temp/robot/helper_tdd_e2e_mock_only_coverage.py

"""Helper script for tdd_e2e_mock_only_coverage.robot smoke tests.

Each subcommand uses AST analysis to inspect the M1-M6 E2E verification
helper files and classify each public test function by its mock and CLI
invocation strategy.  The helper reports the **real** outcome: it exits 0
and prints the sentinel when the assertion holds (bug is fixed), and exits
1 when the bug is still present.  The ``tdd_expected_fail`` tag on the
Robot side handles pass/fail inversion while the bug remains open.

Root cause (Bug #658)
~~~~~~~~~~~~~~~~~~~~~
All 21 CLI-facing tests in ``robot/helper_m*_e2e_verification.py`` use
``unittest.mock.patch`` to replace service factories with ``MagicMock``
objects and invoke the CLI via Typer's in-process ``CliRunner``.  Zero
tests invoke the actual ``agents`` CLI binary via ``subprocess.run``.
"""

from __future__ import annotations

import ast
import sys
from collections.abc import Callable
from dataclasses import dataclass, field
from pathlib import Path

_REPO_ROOT = Path(__file__).resolve().parents[1]

_HELPER_FILES = [
    _REPO_ROOT / f"robot/helper_m{n}_e2e_verification.py" for n in range(1, 7)
]

# Patch targets that indicate service-layer mocking.
_SERVICE_MOCK_INDICATORS: frozenset[str] = frozenset(
    {
        "_get_lifecycle_service",
        "_get_registry_service",
        "_get_namespaced_project_repo",
        "_get_resource_link_repo",
        "_get_resource_registry_service",
        "_get_services",
        "_get_service",
        "get_container",
        "CorrectionService",
    }
)


@dataclass
class FunctionAnalysis:
    """Analysis result for a single function in an E2E helper file."""

    name: str
    suite: str
    uses_cli_runner: bool = False
    uses_mock_patch: bool = False
    uses_subprocess_cli: bool = False
    mock_targets: list[str] = field(default_factory=list)


def _analyze_helper(helper_path: Path) -> list[FunctionAnalysis]:
    """Parse a helper file and classify each top-level function."""
    source = helper_path.read_text(encoding="utf-8")
    tree = ast.parse(source, filename=str(helper_path))
    suite = helper_path.stem

    results: list[FunctionAnalysis] = []

    for node in ast.iter_child_nodes(tree):
        if not isinstance(node, ast.FunctionDef):
            continue
        if node.name.startswith("_"):
            continue

        fa = FunctionAnalysis(name=node.name, suite=suite)

        for child in ast.walk(node):
            if isinstance(child, ast.Attribute):
                attr_name = child.attr
                if attr_name == "invoke" and _is_cli_runner_invoke(child):
                    fa.uses_cli_runner = True
                if attr_name in ("run", "Popen") and _is_subprocess_call(child):
                    fa.uses_subprocess_cli = True

            # Detect ``run_cli(...)`` calls from helper_e2e_common
            # which invoke the real CLI via subprocess.run.
            if isinstance(child, ast.Call) and _is_run_cli_call(child):
                fa.uses_cli_runner = True  # CLI-facing function
                fa.uses_subprocess_cli = True  # via real subprocess

            if isinstance(child, ast.Call) and _is_patch_call(child):
                fa.uses_mock_patch = True
                target = _extract_patch_target(child)
                if target:
                    fa.mock_targets.append(target)

            if isinstance(child, ast.Constant) and isinstance(child.value, str):
                for indicator in _SERVICE_MOCK_INDICATORS:
                    if indicator in child.value:
                        fa.uses_mock_patch = True

        results.append(fa)

    return results


def _is_cli_runner_invoke(node: ast.Attribute) -> bool:
    """Check if attribute access is ``runner.invoke(...)``."""
    return isinstance(node.value, ast.Name) and node.value.id in (
        "runner",
        "cli_runner",
    )


def _is_subprocess_call(node: ast.Attribute) -> bool:
    """Check if attribute access is ``subprocess.run(...)``."""
    return isinstance(node.value, ast.Name) and node.value.id == "subprocess"


def _is_run_cli_call(node: ast.Call) -> bool:
    """Check if a Call node is ``run_cli(...)`` from helper_e2e_common.

    ``run_cli`` is a convenience wrapper around ``subprocess.run`` that
    invokes the real ``agents`` CLI binary.  Functions calling it are
    CLI-facing and exercise subprocess invocation without mocks.
    """
    func = node.func
    return isinstance(func, ast.Name) and func.id == "run_cli"


def _is_patch_call(node: ast.Call) -> bool:
    """Check if a Call node is ``patch(...)`` or ``mock.patch(...)``."""
    func = node.func
    if isinstance(func, ast.Attribute) and func.attr == "patch":
        return True
    return isinstance(func, ast.Name) and func.id == "patch"


def _extract_patch_target(node: ast.Call) -> str | None:
    """Extract the first string argument from a ``patch(...)`` call."""
    if node.args and isinstance(node.args[0], ast.Constant):
        val = node.args[0].value
        if isinstance(val, str):
            return val
    return None


def _get_all_functions() -> list[FunctionAnalysis]:
    """Analyze all M1-M6 helpers and return flat function list."""
    all_functions: list[FunctionAnalysis] = []
    for helper_path in _HELPER_FILES:
        if not helper_path.is_file():
            print(
                f"ERROR: missing helper file {helper_path}",
                file=sys.stderr,
            )
            sys.exit(2)
        all_functions.extend(_analyze_helper(helper_path))
    return all_functions


# ---------------------------------------------------------------------------
# Subcommands
# ---------------------------------------------------------------------------


def check_subprocess_usage() -> None:
    """Assert at least one CLI-facing test uses subprocess.

    Exits 0 with sentinel when the assertion holds (bug fixed).
    Exits 1 when the bug is still present.
    """
    functions = _get_all_functions()
    cli_facing = [f for f in functions if f.uses_cli_runner]
    if not cli_facing:
        print("ERROR: no CLI-facing functions found", file=sys.stderr)
        sys.exit(2)

    subprocess_users = [f for f in cli_facing if f.uses_subprocess_cli]
    if not subprocess_users:
        print(
            f"Bug #658 present: {len(cli_facing)} CLI-facing E2E test "
            f"functions, but NONE use subprocess. All use CliRunner.",
            file=sys.stderr,
        )
        sys.exit(1)

    print("tdd-e2e-subprocess-check-ok")


def check_unmocked_services() -> None:
    """Assert at least one CLI-facing test skips service mocking.

    Exits 0 with sentinel when the assertion holds (bug fixed).
    Exits 1 when the bug is still present.
    """
    functions = _get_all_functions()
    cli_facing = [f for f in functions if f.uses_cli_runner]
    if not cli_facing:
        print("ERROR: no CLI-facing functions found", file=sys.stderr)
        sys.exit(2)

    unmocked = [f for f in cli_facing if not f.uses_mock_patch]
    if not unmocked:
        print(
            f"Bug #658 present: {len(cli_facing)} CLI-facing E2E test "
            f"functions, but ALL mock the service layer.",
            file=sys.stderr,
        )
        sys.exit(1)

    print("tdd-e2e-unmocked-services-ok")


def check_per_suite_coverage() -> None:
    """Assert no suite has 100% mocked CLI tests.

    Exits 0 with sentinel when the assertion holds (bug fixed).
    Exits 1 when the bug is still present.
    """
    per_suite: dict[str, list[FunctionAnalysis]] = {}
    for helper_path in _HELPER_FILES:
        if not helper_path.is_file():
            print(
                f"ERROR: missing helper file {helper_path}",
                file=sys.stderr,
            )
            sys.exit(2)
        suite = helper_path.stem
        per_suite[suite] = _analyze_helper(helper_path)

    all_mocked_suites: list[str] = []
    for suite, functions in per_suite.items():
        cli_facing = [f for f in functions if f.uses_cli_runner]
        if not cli_facing:
            continue
        unmocked = [f for f in cli_facing if not f.uses_mock_patch]
        if not unmocked:
            all_mocked_suites.append(
                f"{suite} ({len(cli_facing)} CLI tests, all mocked)"
            )

    if all_mocked_suites:
        print(
            "Bug #658 present: suites with 100% mocked CLI tests:\n"
            + "\n".join(f"  - {s}" for s in all_mocked_suites),
            file=sys.stderr,
        )
        sys.exit(1)

    print("tdd-e2e-per-suite-coverage-ok")


# ---------------------------------------------------------------------------
# Dispatcher
# ---------------------------------------------------------------------------

_COMMANDS: dict[str, Callable[[], None]] = {
    "check-subprocess-usage": check_subprocess_usage,
    "check-unmocked-services": check_unmocked_services,
    "check-per-suite-coverage": check_per_suite_coverage,
}

if __name__ == "__main__":
    if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
        print(
            f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>",
            file=sys.stderr,
        )
        sys.exit(1)
    cmd = _COMMANDS[sys.argv[1]]
    cmd()