HAL9000/temp
0
0
Fork 0
forked from HAL9000/cleveragents-core
Code Pull Requests Activity
Files
master
temp/robot/helper_m4_e2e_cli.py
T
freemo 5f07316641 fix: wire DI persistence and plan execute/apply for M1 lifecycle 
Fixed 5 bugs preventing the M1 E2E acceptance test from passing:

1. _get_lifecycle_service() in action.py and plan.py bypassed the DI
   container, creating PlanLifecycleService without UnitOfWork. All
   plan/action data was in-memory only and lost between subprocess
   calls. Now uses container.plan_lifecycle_service() for DB persistence.

2. `plan execute` CLI only called service.execute_plan() (a pure state
   transition) without running PlanExecutor phase processing. Rewrote
   to detect the plan's current phase/state and dispatch synchronously:
   Strategize/queued → run_strategize(), Strategize/complete → transition
   + run_execute(), Execute/queued → run_execute().

3. `plan apply` CLI had no plan_id argument. Added optional positional
   plan_id with _lifecycle_apply_with_id() that drives the plan through
   Apply/queued → Apply/processing → Apply/applied.

4. Preflight guardrail in start_strategize() built action_registry from
   the in-memory _actions dict only. Added get_action(plan.action_name)
   call to load the action from DB into cache before the guardrail check.

5. Robot Framework Create File syntax used continuation lines producing
   9 arguments instead of 1. Fixed to use Catenate SEPARATOR=\n then
   pass single variable to Create File. Also fixed --branch main to
   --branch master (git init default).

update mocks for execute_plan CLI changes across unit and integration tests

The new execute_plan() command calls _get_plan_executor() and
service.get_plan(plan_id) for phase/state detection. Existing tests
only mocked _get_lifecycle_service, so MagicMock defaults caused
phase/state comparisons to fail.

Changes across 14 files:
- Patch _get_plan_executor in all test setups that invoke the CLI
  execute command (Behave step files + Robot helper scripts)
- Set service.get_plan.return_value to real Plan objects with correct
  phase/state so the execute_plan dispatch logic works
- Fix error-path tests to use STRATEGIZE/COMPLETE plans so the error
  side_effects are actually reached
- Fix "Multiple plans eligible" → "Multiple plans ready" message text
  to match existing test expectations

increase Robot Framework subprocess timeouts for CI resource contention

Three integration tests were timing out in CI due to resource contention
when pabot runs multiple test suites in parallel. All three pass locally
and the timeouts were simply too tight for constrained CI environments.

- tdd_session_create_di.robot: 30s → 90s (DI container init + DB setup)
- database_integration.robot: 60s → 120s (Run Python Script keyword)
- m3_e2e_verification.robot: 60s → 120s (correction-live-revert spawns
  3 sequential CLI subprocesses with full container initialization)

ISSUES CLOSED: #789
2026-03-15 20:50:02 +00:00

457 lines
17 KiB
Python
Raw Permalink Blame History

"""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 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)."""
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.
"""
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.
"""
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).
"""
# 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")
View Git Blame Copy Permalink