"""Step definitions for TDD Issue #658 — E2E mock-only coverage. These steps use AST analysis to inspect the M1-M6 E2E verification helper files (``robot/helper_m*_e2e_verification.py``) and classify each public test function by its mock and CLI invocation strategy. A function is classified as "CLI-facing" if it uses ``CliRunner`` (from ``typer.testing``) to invoke CLI commands. The test asserts that at least one such CLI-facing function invokes the real ``agents`` CLI via ``subprocess.run`` (or equivalent) without mocking the service layer. With ``@tdd_expected_fail`` the assertions are inverted: the test passes CI while the bug is present (all CLI-facing tests use mocks) and fails once the bug is fixed (signalling that the tag should be removed). Root cause ~~~~~~~~~~ All 21 CLI-facing tests in the M1-M6 E2E helpers use ``unittest.mock.patch`` to replace service factory functions with ``MagicMock`` objects and invoke the CLI via Typer's in-process ``CliRunner``. Zero tests invoke the actual ``agents`` CLI binary via ``subprocess.run``. This prevents the E2E suites from detecting DI wiring bugs, database issues, or process-level failures. """ from __future__ import annotations import ast from dataclasses import dataclass, field from pathlib import Path from typing import Any from behave import given, then, when _REPO_ROOT = Path(__file__).resolve().parents[2] _HELPER_GLOBS = [ "robot/helper_m1_e2e_verification.py", "robot/helper_m2_e2e_verification.py", "robot/helper_m3_e2e_verification.py", "robot/helper_m4_e2e_verification.py", "robot/helper_m5_e2e_verification.py", "robot/helper_m6_e2e_verification.py", ] # 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 # e.g. "helper_m1_e2e_verification" results: list[FunctionAnalysis] = [] for node in ast.iter_child_nodes(tree): if not isinstance(node, ast.FunctionDef): continue # Skip private/dunder helpers and the dispatcher 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 # Detect CliRunner.invoke or runner.invoke if attr_name == "invoke" and _is_cli_runner_invoke(child): fa.uses_cli_runner = True # Detect subprocess.run / subprocess.Popen for CLI 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): # Detect patch target strings that reference service factories 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 an attribute access looks like ``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 an attribute access looks like ``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 # --------------------------------------------------------------------------- # Given steps # --------------------------------------------------------------------------- @given("the M1-M6 E2E verification helper files exist for e2e-mock-audit") def step_helpers_exist(context: Any) -> None: """Verify that all 6 helper files exist on disk.""" missing: list[str] = [] for rel_path in _HELPER_GLOBS: full = _REPO_ROOT / rel_path if not full.is_file(): missing.append(rel_path) assert not missing, f"Missing E2E helper files: {', '.join(missing)}" context.e2e_mock_audit_helpers = [_REPO_ROOT / p for p in _HELPER_GLOBS] # --------------------------------------------------------------------------- # When steps # --------------------------------------------------------------------------- @when("I analyze mock and subprocess usage in M1-M6 helpers for e2e-mock-audit") def step_analyze_all(context: Any) -> None: """Analyze all 6 helpers and store flat list of function analyses.""" all_functions: list[FunctionAnalysis] = [] for helper_path in context.e2e_mock_audit_helpers: all_functions.extend(_analyze_helper(helper_path)) context.e2e_mock_audit_functions = all_functions @when("I analyze mock and subprocess usage per suite for e2e-mock-audit") def step_analyze_per_suite(context: Any) -> None: """Analyze all 6 helpers and store per-suite results.""" per_suite: dict[str, list[FunctionAnalysis]] = {} for helper_path in context.e2e_mock_audit_helpers: suite = helper_path.stem per_suite[suite] = _analyze_helper(helper_path) context.e2e_mock_audit_per_suite = per_suite # --------------------------------------------------------------------------- # Then steps # --------------------------------------------------------------------------- @then( "at least one CLI-facing test should use subprocess instead of " "CliRunner for e2e-mock-audit" ) def step_at_least_one_subprocess(context: Any) -> None: """Assert that at least one CLI-facing function uses subprocess.""" functions: list[FunctionAnalysis] = context.e2e_mock_audit_functions cli_facing = [f for f in functions if f.uses_cli_runner] assert cli_facing, "No CLI-facing functions found at all" subprocess_users = [f for f in cli_facing if f.uses_subprocess_cli] assert subprocess_users, ( f"Bug #658 confirmed: {len(cli_facing)} CLI-facing E2E test " f"functions found across M1-M6, but NONE invoke the real CLI " f"via subprocess. All use Typer's in-process CliRunner.\n" f"CLI-facing functions: " f"{', '.join(f'{f.suite}::{f.name}' for f in cli_facing)}" ) @then( "at least one CLI-facing test should not mock the service factory " "for e2e-mock-audit" ) def step_at_least_one_unmocked(context: Any) -> None: """Assert that at least one CLI-facing function skips service mocking.""" functions: list[FunctionAnalysis] = context.e2e_mock_audit_functions cli_facing = [f for f in functions if f.uses_cli_runner] assert cli_facing, "No CLI-facing functions found at all" unmocked = [f for f in cli_facing if not f.uses_mock_patch] assert unmocked, ( f"Bug #658 confirmed: {len(cli_facing)} CLI-facing E2E test " f"functions found across M1-M6, but ALL mock the service layer " f"via unittest.mock.patch. None exercise the real DI container.\n" f"Mocked functions: " f"{', '.join(f'{f.suite}::{f.name}' for f in cli_facing)}" ) @then( "every suite with CLI tests should have at least one unmocked test " "for e2e-mock-audit" ) def step_every_suite_has_unmocked(context: Any) -> None: """Assert that no suite has 100% mocked CLI tests.""" per_suite: dict[str, list[FunctionAnalysis]] = context.e2e_mock_audit_per_suite 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 # Suite has no CLI tests — skip (e.g. M5) 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)" ) assert not all_mocked_suites, ( "Bug #658 confirmed: the following suites have 100% mocked CLI " "tests with no real code path coverage:\n" + "\n".join(f" - {s}" for s in all_mocked_suites) )