cleveragents-core/robot/helper_scale_test.py

"""Robot Framework helper for scale test fixture validation.

Provides a CLI-style interface for Robot to invoke fixture loading,
metadata validation, threshold checks, and profile simulation.
Exit code 0 = success, 1 = failure.

Usage:
    python robot/helper_scale_test.py load-metadata
    python robot/helper_scale_test.py validate-profiles
    python robot/helper_scale_test.py validate-thresholds
    python robot/helper_scale_test.py check-monotonicity
    python robot/helper_scale_test.py simulate-distribution
    python robot/helper_scale_test.py check-generator-docs
"""

from __future__ import annotations

import json
import math
import sys
from pathlib import Path

# Ensure the src directory is on the import path.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)

_FIXTURES_DIR = Path(__file__).resolve().parents[1] / "features" / "fixtures" / "scale"


def _load_json(filename: str) -> dict:
    """Load a JSON fixture file."""
    path = _FIXTURES_DIR / filename
    if not path.exists():
        print(f"ERROR: fixture not found: {path}")
        sys.exit(1)
    with open(path) as fh:
        return json.load(fh)


def _cmd_load_metadata() -> int:
    """Load scale_metadata.json and verify basic structure."""
    data = _load_json("scale_metadata.json")
    assert "schema_version" in data, "Missing schema_version"
    assert "scale_profiles" in data, "Missing scale_profiles"
    assert len(data["scale_profiles"]) > 0, "No profiles defined"
    print(f"scale-metadata-ok: {len(data['scale_profiles'])} profiles loaded")
    return 0


def _cmd_validate_profiles() -> int:
    """Validate all scale profiles have required fields."""
    data = _load_json("scale_metadata.json")
    required_fields = [
        "name",
        "file_count",
        "total_size_mb",
        "language_mix",
        "expected_index_time_s",
        "expected_decomposition_time_s",
    ]
    expected_names = {"small", "medium", "large", "xlarge", "xxlarge"}
    actual_names = {p["name"] for p in data["scale_profiles"]}
    assert actual_names == expected_names, (
        f"Expected profiles {expected_names}, got {actual_names}"
    )
    for profile in data["scale_profiles"]:
        for field in required_fields:
            assert field in profile, f"Missing {field} in profile {profile.get('name')}"
        # Verify language mix sums to 1.0
        mix_total = sum(profile["language_mix"].values())
        assert math.isclose(mix_total, 1.0, rel_tol=1e-6), (
            f"Language mix for {profile['name']} sums to {mix_total}"
        )
    print(f"scale-profiles-ok: {len(data['scale_profiles'])} profiles validated")
    return 0


def _cmd_validate_thresholds() -> int:
    """Load and validate baseline thresholds structure."""
    data = _load_json("baseline_thresholds.json")
    assert "schema_version" in data, "Missing schema_version"
    for section in ("indexing", "decomposition"):
        assert section in data, f"Missing section: {section}"
        for file_count in (
            "1000_files",
            "5000_files",
            "10000_files",
            "50000_files",
            "100000_files",
        ):
            assert file_count in data[section], f"Missing {file_count} in {section}"
            for pct in ("p50_ms", "p95_ms", "p99_ms"):
                assert pct in data[section][file_count], (
                    f"Missing {pct} in {section}[{file_count}]"
                )
    print("scale-thresholds-ok: all threshold sections validated")
    return 0


def _cmd_check_monotonicity() -> int:
    """Verify p50 < p95 < p99 for all threshold entries."""
    data = _load_json("baseline_thresholds.json")
    for section in ("indexing", "decomposition"):
        for file_count, thresholds in data[section].items():
            p50 = thresholds["p50_ms"]
            p95 = thresholds["p95_ms"]
            p99 = thresholds["p99_ms"]
            assert p50 < p95 < p99, (
                f"Non-monotonic in {section}[{file_count}]: "
                f"p50={p50}, p95={p95}, p99={p99}"
            )
    print("scale-monotonicity-ok: all thresholds are monotonically increasing")
    return 0


def _cmd_simulate_distribution() -> int:
    """Generate file distributions for all profiles and verify totals."""
    data = _load_json("scale_metadata.json")
    for profile in data["scale_profiles"]:
        total = profile["file_count"]
        mix = profile["language_mix"]
        distribution: dict[str, int] = {}
        allocated = 0
        items = sorted(mix.items(), key=lambda x: x[1], reverse=True)
        for i, (lang, ratio) in enumerate(items):
            if i == len(items) - 1:
                distribution[lang] = total - allocated
            else:
                count = round(total * ratio)
                distribution[lang] = count
                allocated += count

        actual_total = sum(distribution.values())
        assert actual_total == total, (
            f"Profile {profile['name']}: expected {total} files, got {actual_total}"
        )
        # Verify distribution roughly matches mix
        for lang, expected_ratio in mix.items():
            actual_ratio = distribution[lang] / total
            assert abs(actual_ratio - expected_ratio) < 0.05, (
                f"Profile {profile['name']}, lang {lang}: "
                f"expected ~{expected_ratio}, got {actual_ratio}"
            )
        print(f"  {profile['name']}: {distribution}")

    print("scale-distribution-ok: all profile distributions validated")
    return 0


def _cmd_check_generator_docs() -> int:
    """Verify the generator instructions document exists and has content."""
    path = _FIXTURES_DIR / "generator_instructions.md"
    assert path.exists(), f"Generator instructions not found: {path}"
    content = path.read_text()
    assert len(content) > 100, "Generator instructions file is too short"
    assert "## Overview" in content, "Missing Overview section"
    assert "## Generating" in content or "## Prerequisites" in content, (
        "Missing generation instructions"
    )
    print("scale-generator-docs-ok: generator instructions validated")
    return 0


def main() -> int:
    """Entry point called by Robot Framework ``Run Process``."""
    if len(sys.argv) < 2:
        print(
            "Usage: helper_scale_test.py "
            "<load-metadata|validate-profiles|validate-thresholds|"
            "check-monotonicity|simulate-distribution|check-generator-docs>"
        )
        return 1

    command = sys.argv[1]

    commands = {
        "load-metadata": _cmd_load_metadata,
        "validate-profiles": _cmd_validate_profiles,
        "validate-thresholds": _cmd_validate_thresholds,
        "check-monotonicity": _cmd_check_monotonicity,
        "simulate-distribution": _cmd_simulate_distribution,
        "check-generator-docs": _cmd_check_generator_docs,
    }

    handler = commands.get(command)
    if handler is None:
        print(f"Unknown command: {command}")
        return 1

    try:
        return handler()
    except (AssertionError, KeyError, json.JSONDecodeError) as exc:
        print(f"FAIL: {exc}")
        return 1


if __name__ == "__main__":
    sys.exit(main())