"""M4 E2E CLI integration tests (happy-path). Tests the ``agents plan`` CLI subcommands (diff, use, execute, tree) by invoking them via Typer's ``CliRunner`` with mocked services. Error-path tests live in ``helper_m4_e2e_cli_errors`` to keep both files under the 500-line CONTRIBUTING.md limit. Usage (via dispatcher): python robot/helper_m4_e2e_verification.py plan-diff python robot/helper_m4_e2e_verification.py cli-plan-use python robot/helper_m4_e2e_verification.py cli-plan-execute python robot/helper_m4_e2e_verification.py cli-plan-tree """ from __future__ import annotations import json import re import sys from pathlib import Path from unittest.mock import MagicMock, patch # 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.append(_SRC) from helper_m4_e2e_common import ( # noqa: E402 _CHILD_A_ULID, _CHILD_B_ULID, _ROOT_ULID, _assert_exit_code, _assert_mock_called_once, _assert_mock_called_once_with, _fail, _make_subplan_statuses, _mock_parent_plan, ) from helpers_common import reset_global_state # noqa: E402 from typer.testing import CliRunner # noqa: E402 from cleveragents.cli.commands.plan import app as plan_app # noqa: E402 from cleveragents.domain.models.core.decision import ( # noqa: E402 Decision, DecisionType, ) from cleveragents.domain.models.core.plan import ( # noqa: E402 NamespacedName, PlanPhase, ProcessingState, ProjectLink, ) _runner = CliRunner() # Decision ULIDs with descriptive names indicating their decision type. _DECISION_PROMPT_DEF = "01KHDE6WWS2171PWW3GJEBXZ01" _DECISION_STRATEGY = "01KHDE6WWS2171PWW3GJEBXZ02" _DECISION_PARALLEL_SPAWN = "01KHDE6WWS2171PWW3GJEBXZ03" _DECISION_SPAWN_API = "01KHDE6WWS2171PWW3GJEBXZ04" _DECISION_SPAWN_UI = "01KHDE6WWS2171PWW3GJEBXZ05" # --------------------------------------------------------------------------- # plan-diff # --------------------------------------------------------------------------- def plan_diff() -> None: """Verify merged results via ``agents plan diff`` (mocked service).""" reset_global_state() mock_apply_svc = MagicMock() mock_apply_svc.diff.return_value = ( "--- a/src/api.py\n" "+++ b/src/api.py\n" "@@ -1,3 +1,5 @@\n" " # API module\n" "+from fastapi import FastAPI\n" "+app = FastAPI()\n" " # existing code\n" "--- a/src/ui.py\n" "+++ b/src/ui.py\n" "@@ -1,2 +1,3 @@\n" " # UI module\n" "+import react\n" ) with patch( "cleveragents.cli.commands.plan._get_apply_service", return_value=mock_apply_svc, ): result = _runner.invoke(plan_app, ["diff", _ROOT_ULID]) _assert_exit_code(result, "plan diff") # Verify diff was called with the plan_id (kwargs + positional # fallback to avoid IndexError if called as keyword argument). _assert_mock_called_once(mock_apply_svc.diff, "diff") call_args = mock_apply_svc.diff.call_args plan_id_arg = call_args.kwargs.get("plan_id") if plan_id_arg is None and call_args.args: plan_id_arg = call_args.args[0] if plan_id_arg != _ROOT_ULID: _fail(f"diff called with wrong plan_id: {plan_id_arg}") # Verify the ``fmt`` kwarg was forwarded. The production CLI # calls ``service.diff(plan_id, fmt=fmt)``; a regression that # dropped the ``fmt`` argument must be detected. The default # value is ``"rich"`` (from the CLI ``--format`` option), so # both presence and value are asserted strictly. fmt_arg = call_args.kwargs.get("fmt") if fmt_arg is None and len(call_args.args) > 1: fmt_arg = call_args.args[1] if fmt_arg is None: _fail( "diff was not called with 'fmt' argument — " "expected fmt='rich' (default)" ) if fmt_arg != "rich": _fail(f"diff called with unexpected fmt: {fmt_arg!r} — expected 'rich'") # Verify diff content is rendered in CLI output (M4 criterion: # "plan diff shows merged results"). if "api.py" not in result.output: _fail( f"diff output should contain 'api.py' from merged results\n" f"output={result.output}" ) if "ui.py" not in result.output: _fail( f"diff output should contain 'ui.py' from merged results\n" f"output={result.output}" ) print("m4-plan-diff-ok") # --------------------------------------------------------------------------- # cli-plan-use # --------------------------------------------------------------------------- def cli_plan_use() -> None: """Verify ``agents plan use`` CLI creates a plan with subplan config. Mocks the lifecycle service and invokes the actual CLI command via Typer's CliRunner. Asserts the service receives the correct action name and project, and the CLI exits cleanly. """ reset_global_state() mock_service = MagicMock() # Mock action lookup -- use_action needs get_action_by_name first mock_action = MagicMock() mock_action.namespaced_name = NamespacedName.parse("local/refactor-action") mock_service.get_action_by_name.return_value = mock_action # Mock use_action to return a plan in Strategize phase plan = _mock_parent_plan( phase=PlanPhase.STRATEGIZE, state=ProcessingState.QUEUED, ) mock_service.use_action.return_value = plan with patch( "cleveragents.cli.commands.plan._get_lifecycle_service", return_value=mock_service, ): result = _runner.invoke( plan_app, ["use", "local/refactor-action", "local/monorepo"], ) _assert_exit_code(result, "plan use") # Verify the action was looked up _assert_mock_called_once_with( mock_service.get_action_by_name, "get_action_by_name", "local/refactor-action", ) # Verify use_action was called with correct action and project _assert_mock_called_once(mock_service.use_action, "use_action") call_kwargs = mock_service.use_action.call_args # action_name: kwargs with positional fallback action_arg = call_kwargs.kwargs.get("action_name") if action_arg is None and call_kwargs.args: action_arg = call_kwargs.args[0] if action_arg != "local/refactor-action": _fail(f"use_action called with action={action_arg}") # project_links: kwargs with positional fallback (symmetric # with action_name to avoid asymmetric arg extraction). project_links = call_kwargs.kwargs.get("project_links") if project_links is None and len(call_kwargs.args) > 1: project_links = call_kwargs.args[1] if project_links is None: _fail("use_action not called with project_links argument") # Type guard: verify all items are ProjectLink instances if not all(isinstance(pl, ProjectLink) for pl in project_links): _fail( f"project_links should be list of ProjectLink, " f"got {[type(p).__name__ for p in project_links]}" ) project_names = [pl.project_name for pl in project_links] if "local/monorepo" not in project_names: _fail(f"use_action project_links missing 'local/monorepo': {project_names}") print("m4-cli-plan-use-ok") # --------------------------------------------------------------------------- # cli-plan-execute # --------------------------------------------------------------------------- def cli_plan_execute() -> None: """Verify ``agents plan execute`` CLI transitions plan to Execute. Mocks the lifecycle service and invokes the actual CLI command. Asserts the plan transitions to Execute phase and the CLI renders the plan details. """ reset_global_state() mock_service = MagicMock() # get_plan is called for the read-only guard pre_plan = _mock_parent_plan( phase=PlanPhase.STRATEGIZE, state=ProcessingState.COMPLETE, ) mock_service.get_plan.return_value = pre_plan # execute_plan returns the plan in Execute phase with subplans statuses = _make_subplan_statuses() post_plan = _mock_parent_plan( phase=PlanPhase.EXECUTE, state=ProcessingState.PROCESSING, subplan_statuses=statuses, ) mock_service.execute_plan.return_value = post_plan with ( patch( "cleveragents.cli.commands.plan._get_lifecycle_service", return_value=mock_service, ), patch( "cleveragents.cli.commands.plan._get_plan_executor", return_value=MagicMock(), ), ): result = _runner.invoke(plan_app, ["execute", _ROOT_ULID]) _assert_exit_code(result, "plan execute") # Verify execute_plan was called with the plan ID _assert_mock_called_once_with( mock_service.execute_plan, "execute_plan", _ROOT_ULID ) # Verify CLI output references the plan (rich rendering) if _ROOT_ULID not in result.output: _fail( f"CLI output should contain plan ID {_ROOT_ULID}\n" f"output={result.output}" ) # Verify CLI output reflects the Execute phase transition. # Use word-boundary regex to match "execute" as a distinct word, # avoiding false positives from incidental substrings. if not re.search(r"\bexecute\b", result.output, re.IGNORECASE): _fail(f"CLI output should mention 'execute' phase\noutput={result.output}") # NOTE: Domain-level post_plan assertions (phase, subplan_statuses, # has_subplans) that were previously here have been removed. They # verified values the test itself constructed via _mock_parent_plan, # constituting a tautology that could never fail. The CLI does not # render subplan_count or has_subplans in its output, so these # properties cannot be meaningfully verified from CLI output alone. # Domain-level coverage for these properties is provided by the # spawn-subplans and parent-tracking domain tests. print("m4-cli-plan-execute-ok") # --------------------------------------------------------------------------- # cli-plan-tree # --------------------------------------------------------------------------- def cli_plan_tree() -> None: """Verify ``agents plan tree`` CLI displays subplan hierarchy. Creates real Decision objects forming a subplan tree: - prompt_definition (root) - strategy_choice - subplan_parallel_spawn - subplan_spawn (child A) - subplan_spawn (child B) Mocks the DI container and DecisionService, then invokes the CLI command with ``--format json`` and verifies the JSON output contains the expected decision types and hierarchy. NOTE: This test patches ``get_container`` at its definition site rather than ``_get_lifecycle_service`` because the production ``tree`` command resolves the ``DecisionService`` directly from the DI container (not via the lifecycle service helper). """ reset_global_state() # Build a realistic decision tree with subplan decisions decisions = [ Decision( decision_id=_DECISION_PROMPT_DEF, plan_id=_ROOT_ULID, parent_decision_id=None, sequence_number=0, decision_type=DecisionType.PROMPT_DEFINITION, question="What is the plan prompt?", chosen_option="Refactor code across monorepo modules", confidence_score=1.0, ), Decision( decision_id=_DECISION_STRATEGY, plan_id=_ROOT_ULID, parent_decision_id=_DECISION_PROMPT_DEF, sequence_number=1, decision_type=DecisionType.STRATEGY_CHOICE, question="How to decompose the refactoring?", chosen_option="Parallel subplans per module", confidence_score=0.85, ), Decision( decision_id=_DECISION_PARALLEL_SPAWN, plan_id=_ROOT_ULID, parent_decision_id=_DECISION_STRATEGY, sequence_number=2, decision_type=DecisionType.SUBPLAN_PARALLEL_SPAWN, question="Which subplans to run in parallel?", chosen_option="Spawn API and UI subplans concurrently", downstream_plan_ids=[_CHILD_A_ULID, _CHILD_B_ULID], ), Decision( decision_id=_DECISION_SPAWN_API, plan_id=_ROOT_ULID, parent_decision_id=_DECISION_PARALLEL_SPAWN, sequence_number=3, decision_type=DecisionType.SUBPLAN_SPAWN, question="Spawn subplan for API module?", chosen_option="Spawn API refactor subplan", downstream_plan_ids=[_CHILD_A_ULID], ), Decision( decision_id=_DECISION_SPAWN_UI, plan_id=_ROOT_ULID, parent_decision_id=_DECISION_PARALLEL_SPAWN, sequence_number=4, decision_type=DecisionType.SUBPLAN_SPAWN, question="Spawn subplan for UI module?", chosen_option="Spawn UI refactor subplan", downstream_plan_ids=[_CHILD_B_ULID], ), ] # Mock the DI container and DecisionService. # The production ``tree`` command calls ``container.decision_service()`` # (not ``container.resolve()``), so the mock must match that API. mock_container = MagicMock() mock_decision_svc = MagicMock() mock_decision_svc.list_decisions.return_value = decisions mock_container.decision_service.return_value = mock_decision_svc with patch( "cleveragents.application.container.get_container", return_value=mock_container, ): result = _runner.invoke( plan_app, ["tree", _ROOT_ULID, "--format", "json"], ) _assert_exit_code(result, "plan tree") # Verify the decision service was queried _assert_mock_called_once_with( mock_decision_svc.list_decisions, "list_decisions", _ROOT_ULID, ) # Parse the JSON output and verify tree structure try: tree_data: list[dict[str, object]] = json.loads(result.output) except json.JSONDecodeError: _fail(f"plan tree output is not valid JSON:\n{result.output}") return # unreachable; _fail raises SystemExit # Verify the tree is a list with exactly 1 root node if not isinstance(tree_data, list): _fail(f"tree output should be a list, got {type(tree_data).__name__}") if len(tree_data) != 1: _fail(f"tree should have 1 root, got {len(tree_data)}") # Root must be prompt_definition root = tree_data[0] if root.get("type") != "prompt_definition": _fail(f"root type should be prompt_definition, got {root.get('type')}") if root.get("decision_id") != _DECISION_PROMPT_DEF: _fail(f"root decision_id mismatch: {root.get('decision_id')}") # Root -> strategy_choice root_children = root.get("children", []) if not isinstance(root_children, list) or len(root_children) != 1: _fail(f"root should have 1 child, got {root_children!r}") strategy_node = root_children[0] if strategy_node.get("type") != "strategy_choice": _fail( f"root child should be strategy_choice, got {strategy_node.get('type')}" ) # strategy_choice -> subplan_parallel_spawn strategy_children = strategy_node.get("children", []) if not isinstance(strategy_children, list) or len(strategy_children) != 1: _fail(f"strategy_choice should have 1 child, got {strategy_children!r}") parallel_node = strategy_children[0] if parallel_node.get("type") != "subplan_parallel_spawn": _fail( f"strategy child should be subplan_parallel_spawn, " f"got {parallel_node.get('type')}" ) # subplan_parallel_spawn -> 2x subplan_spawn spawn_children = parallel_node.get("children", []) if not isinstance(spawn_children, list) or len(spawn_children) != 2: _fail( f"subplan_parallel_spawn should have 2 children, got {spawn_children!r}" ) for sc in spawn_children: if sc.get("type") != "subplan_spawn": _fail( f"parallel spawn child should be subplan_spawn, " f"got {sc.get('type')}" ) # Verify total node count: 5 decisions in the tree def _count_nodes(nodes: list[dict[str, object]]) -> int: total = 0 for n in nodes: total += 1 children = n.get("children") if isinstance(children, list): total += _count_nodes(children) return total node_count = _count_nodes(tree_data) if node_count != 5: _fail(f"tree should have 5 nodes, got {node_count}") print("m4-cli-plan-tree-ok")