"""Step definitions for plan_cli_commands_r2.feature. Targets remaining partial branches in ``cleveragents.cli.commands.plan`` (plan.py) - round 2, split 2 of 3. Covers: - ``use_action`` argument parsing: int/float/bool/string, missing '=', invalid automation profile, invalid actor overrides - ``execute_plan`` auto-resolve: 0 plans / >1 plans / exactly 1 - ``apply_plan`` auto-resolve: same subcases - ``list_plans``: invalid phase/state, empty result, project truncation - ``revert_plan``: invalid to-phase, non-rich format - ``plan_status``: no plans, non-rich list, non-rich single plan - ``correct_decision``: invalid mode, empty guidance - ``plan_diff``: correction flag All step text uses the ``r2plan-`` prefix to avoid collisions. """ from __future__ import annotations from datetime import datetime from typing import Any from unittest.mock import MagicMock, patch from behave import given, then, when from typer.testing import CliRunner from cleveragents.cli.commands import plan as plan_module from cleveragents.cli.commands.plan import app as plan_app from cleveragents.domain.models.core.plan import ( AutomationProfileRef, NamespacedName, Plan, PlanIdentity, PlanInvariant, PlanPhase, PlanTimestamps, ProcessingState, ProjectLink, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- _ULID_BASE = "01ARZ3NDEKTSV4RRFFQ69G5F" _runner = CliRunner() def _ulid(suffix: str = "A1") -> str: """Return a valid 26-char ULID for tests. ULIDs use Crockford's Base32 (0-9, A-H, J-K, M-N, P-T, V-Z; no I/L/O/U). """ # Map potentially invalid chars to valid Crockford Base32 cleaned = ( suffix.replace("I", "J").replace("L", "K").replace("O", "P").replace("U", "V") ) base = _ULID_BASE + cleaned return base[:26] def _make_plan( *, plan_id: str | None = None, name: str = "local/r2-plan", description: str = "Test plan for r2 coverage", phase: PlanPhase = PlanPhase.STRATEGIZE, processing_state: ProcessingState = ProcessingState.QUEUED, project_links: list[ProjectLink] | None = None, automation_profile: AutomationProfileRef | None = None, invariants: list[PlanInvariant] | None = None, validation_summary: dict[str, Any] | None = None, error_message: str | None = None, last_completed_step: int = -1, last_checkpoint_id: str | None = None, definition_of_done: str | None = None, arguments: dict[str, Any] | None = None, arguments_order: list[str] | None = None, estimation_actor: str | None = None, invariant_actor: str | None = None, timestamps: PlanTimestamps | None = None, action_name: str = "local/test-action", ) -> Plan: if timestamps is None: timestamps = PlanTimestamps( created_at=datetime.now(), updated_at=datetime.now(), ) return Plan( identity=PlanIdentity(plan_id=plan_id or _ulid("A1")), namespaced_name=NamespacedName.parse(name), action_name=action_name, description=description, definition_of_done=definition_of_done, phase=phase, processing_state=processing_state, strategy_actor=None, execution_actor=None, project_links=project_links or [], automation_profile=automation_profile, invariants=invariants or [], validation_summary=validation_summary, error_message=error_message, last_completed_step=last_completed_step, last_checkpoint_id=last_checkpoint_id, arguments=arguments or {}, arguments_order=arguments_order or [], estimation_actor=estimation_actor, invariant_actor=invariant_actor, timestamps=timestamps, created_by=None, reusable=True, read_only=False, ) # --------------------------------------------------------------------------- # Given steps - mocked lifecycle service (shared for CLI command scenarios) # --------------------------------------------------------------------------- @given("r2plan-a mocked lifecycle service") def step_mocked_lifecycle(context: Any) -> None: context.r2_mock_svc = MagicMock() context.r2_cleanups = [] # list[Any] # Patch the lifecycle service getter p = patch( "cleveragents.cli.commands.plan._get_lifecycle_service", return_value=context.r2_mock_svc, ) p.start() context.r2_cleanups.append(p.stop) # Patch the plan executor getter (needed by execute_plan command) p2 = patch( "cleveragents.cli.commands.plan._get_plan_executor", return_value=MagicMock(), ) p2.start() context.r2_cleanups.append(p2.stop) # Replace the module-level console with a wider one for table tests from rich.console import Console as RichConsole wide_console = RichConsole(width=200) original_console = plan_module.console plan_module.console = wide_console def _restore_console() -> None: plan_module.console = original_console context.r2_cleanups.append(_restore_console) # Default: get_action_by_name returns a mock action action = MagicMock() action.namespaced_name = "local/test" context.r2_mock_svc.get_action_by_name.return_value = action # Default: use_action returns a plan (needed for use_action scenarios) context.r2_use_plan = _make_plan() context.r2_mock_svc.use_action.return_value = context.r2_use_plan # Register cleanup if not hasattr(context, "_r2_after_scenario"): def _cleanup(ctx: Any) -> None: for c in getattr(ctx, "r2_cleanups", []): c() context.add_cleanup(_cleanup, context) @given("r2plan-the service lists no complete strategize plans") def step_no_strategize_plans(context: Any) -> None: context.r2_mock_svc.list_plans.return_value = [] @given("r2plan-the service lists multiple complete strategize plans") def step_multi_strategize_plans(context: Any) -> None: p1 = _make_plan( plan_id=_ulid("B1"), processing_state=ProcessingState.COMPLETE, ) p2 = _make_plan( plan_id=_ulid("B2"), processing_state=ProcessingState.COMPLETE, ) context.r2_mock_svc.list_plans.return_value = [p1, p2] @given("r2plan-the service lists exactly one complete strategize plan") def step_one_strategize_plan(context: Any) -> None: p = _make_plan( plan_id=_ulid("C1"), processing_state=ProcessingState.COMPLETE, ) context.r2_mock_svc.list_plans.return_value = [p] context.r2_mock_svc.get_plan.return_value = p context.r2_mock_svc.execute_plan.return_value = p @given("r2plan-the service lists no complete execute plans") def step_no_execute_plans(context: Any) -> None: context.r2_mock_svc.list_plans.return_value = [] @given("r2plan-the service lists multiple complete execute plans") def step_multi_execute_plans(context: Any) -> None: p1 = _make_plan( plan_id=_ulid("D1"), phase=PlanPhase.EXECUTE, processing_state=ProcessingState.COMPLETE, ) p2 = _make_plan( plan_id=_ulid("D2"), phase=PlanPhase.EXECUTE, processing_state=ProcessingState.COMPLETE, ) context.r2_mock_svc.list_plans.return_value = [p1, p2] @given("r2plan-the service lists exactly one complete execute plan") def step_one_execute_plan(context: Any) -> None: p = _make_plan( plan_id=_ulid("E1"), phase=PlanPhase.EXECUTE, processing_state=ProcessingState.COMPLETE, ) context.r2_mock_svc.list_plans.return_value = [p] context.r2_mock_svc.apply_plan.return_value = p @given("r2plan-the service lists no plans") def step_no_plans(context: Any) -> None: context.r2_mock_svc.list_plans.return_value = [] @given("r2plan-the service lists some plans") def step_some_plans(context: Any) -> None: plans = [_make_plan(plan_id=_ulid("F1"))] context.r2_mock_svc.list_plans.return_value = plans @given("r2plan-the service can get a plan by id") def step_get_plan_by_id(context: Any) -> None: plan = _make_plan(plan_id=_ulid("G1")) context.r2_mock_svc.get_plan.return_value = plan context.r2_plan_id = _ulid("G1") @given("r2plan-the service can revert a plan") def step_service_can_revert(context: Any) -> None: plan = _make_plan(plan_id=_ulid("H1")) context.r2_mock_svc.revert_plan.return_value = plan @given("r2plan-the service lists a plan with 4 project links") def step_plan_with_4_links(context: Any) -> None: links = [ProjectLink(project_name=f"local/proj-{i}") for i in range(4)] p = _make_plan(plan_id=_ulid("I1"), project_links=links) context.r2_mock_svc.list_plans.return_value = [p] # --------------------------------------------------------------------------- # When steps - CLI use_action # --------------------------------------------------------------------------- @when('r2plan-I invoke use with action "{action}" and arg "{arg_str}"') def step_invoke_use_with_arg(context: Any, action: str, arg_str: str) -> None: context.r2_result = _runner.invoke( plan_app, ["use", action, "--arg", arg_str], ) # Capture the arguments passed to use_action for inspection if context.r2_mock_svc.use_action.called: call_kwargs = context.r2_mock_svc.use_action.call_args context.r2_parsed_args = ( call_kwargs.kwargs.get("arguments", call_kwargs[1].get("arguments", {})) if call_kwargs else {} ) else: context.r2_parsed_args = {} @when('r2plan-I invoke use with action "{action}" and automation profile "{profile}"') def step_invoke_use_with_profile(context: Any, action: str, profile: str) -> None: context.r2_result = _runner.invoke( plan_app, ["use", action, "--automation-profile", profile], ) @when('r2plan-I invoke use with action "{action}" and strategy actor "{actor}"') def step_invoke_use_strategy_actor(context: Any, action: str, actor: str) -> None: context.r2_result = _runner.invoke( plan_app, ["use", action, "--strategy-actor", actor], ) @when('r2plan-I invoke use with action "{action}" and execution actor "{actor}"') def step_invoke_use_execution_actor(context: Any, action: str, actor: str) -> None: context.r2_result = _runner.invoke( plan_app, ["use", action, "--execution-actor", actor], ) # --------------------------------------------------------------------------- # When steps - CLI execute / apply # --------------------------------------------------------------------------- @when("r2plan-I invoke execute without plan_id") def step_invoke_execute_no_id(context: Any) -> None: context.r2_result = _runner.invoke(plan_app, ["execute"]) @when("r2plan-I invoke apply without plan_id") def step_invoke_apply_no_id(context: Any) -> None: context.r2_result = _runner.invoke(plan_app, ["apply", "--yes"]) # --------------------------------------------------------------------------- # When steps - CLI list # --------------------------------------------------------------------------- @when('r2plan-I invoke list with phase "{phase}"') def step_invoke_list_phase(context: Any, phase: str) -> None: context.r2_result = _runner.invoke(plan_app, ["list", "--phase", phase]) @when('r2plan-I invoke list with state "{state}"') def step_invoke_list_state(context: Any, state: str) -> None: context.r2_result = _runner.invoke(plan_app, ["list", "--state", state]) @when("r2plan-I invoke list") def step_invoke_list(context: Any) -> None: context.r2_result = _runner.invoke(plan_app, ["list"]) # --------------------------------------------------------------------------- # When steps - CLI revert # --------------------------------------------------------------------------- @when('r2plan-I invoke revert with plan "{pid}" and invalid phase "{phase}"') def step_invoke_revert_invalid(context: Any, pid: str, phase: str) -> None: context.r2_result = _runner.invoke(plan_app, ["revert", pid, "--to-phase", phase]) @when('r2plan-I invoke revert with plan "{pid}" and format "{fmt}"') def step_invoke_revert_json(context: Any, pid: str, fmt: str) -> None: context.r2_result = _runner.invoke(plan_app, ["revert", pid, "--format", fmt]) # --------------------------------------------------------------------------- # When steps - CLI status # --------------------------------------------------------------------------- @when("r2plan-I invoke status without plan_id") def step_invoke_status_no_id(context: Any) -> None: context.r2_result = _runner.invoke(plan_app, ["status"]) @when('r2plan-I invoke status without plan_id and format "{fmt}"') def step_invoke_status_no_id_fmt(context: Any, fmt: str) -> None: context.r2_result = _runner.invoke(plan_app, ["status", "--format", fmt]) @when('r2plan-I invoke status with plan_id and format "{fmt}"') def step_invoke_status_with_id_fmt(context: Any, fmt: str) -> None: pid = getattr(context, "r2_plan_id", _ulid("G1")) context.r2_result = _runner.invoke(plan_app, ["status", pid, "--format", fmt]) # --------------------------------------------------------------------------- # When steps - CLI correct # --------------------------------------------------------------------------- @when('r2plan-I invoke correct with mode "{mode}"') def step_invoke_correct_bad_mode(context: Any, mode: str) -> None: context.r2_result = _runner.invoke( plan_app, ["correct", "DEC-001", "--mode", mode, "--guidance", "fix it", "--yes"], ) @when("r2plan-I invoke correct with empty guidance") def step_invoke_correct_empty_guidance(context: Any) -> None: context.r2_result = _runner.invoke( plan_app, ["correct", "DEC-001", "--mode", "revert", "--guidance", "", "--yes"], ) # --------------------------------------------------------------------------- # When steps - CLI diff # --------------------------------------------------------------------------- @when('r2plan-I invoke diff with plan "{pid}" and correction "{corr}"') def step_invoke_diff_correction(context: Any, pid: str, corr: str) -> None: context.r2_result = _runner.invoke( plan_app, ["diff", pid, "--correction", corr], ) # --------------------------------------------------------------------------- # Then steps - CLI command assertions # --------------------------------------------------------------------------- @then("r2plan-the command should abort") def step_cmd_abort(context: Any) -> None: assert context.r2_result.exit_code != 0, ( f"Expected abort but got exit code {context.r2_result.exit_code}.\n" f"Output: {context.r2_result.output}" ) @then("r2plan-the command should succeed") def step_cmd_succeed(context: Any) -> None: assert context.r2_result.exit_code == 0, ( f"Expected success but got exit code {context.r2_result.exit_code}.\n" f"Output: {context.r2_result.output}" ) @then('r2plan-the output should contain "{text}"') def step_cli_output_contains(context: Any, text: str) -> None: output = context.r2_result.output assert text in output, f"Expected '{text}' in output:\n{output}" # -- use_action argument parsing assertions -- @then('r2plan-the parsed arguments should have "{key}" as int {val:d}') def step_arg_int(context: Any, key: str, val: int) -> None: call_args = context.r2_mock_svc.use_action.call_args args = call_args.kwargs.get("arguments") or call_args[1].get("arguments") or {} assert args.get(key) == val, f"Expected {key}={val}, got {args.get(key)}" assert isinstance(args[key], int) @then('r2plan-the parsed arguments should have "{key}" as float {val:g}') def step_arg_float(context: Any, key: str, val: float) -> None: call_args = context.r2_mock_svc.use_action.call_args args = call_args.kwargs.get("arguments") or call_args[1].get("arguments") or {} assert args.get(key) == val, f"Expected {key}={val}, got {args.get(key)}" assert isinstance(args[key], float) @then('r2plan-the parsed arguments should have "{key}" as bool true') def step_arg_bool_true(context: Any, key: str) -> None: call_args = context.r2_mock_svc.use_action.call_args args = call_args.kwargs.get("arguments") or call_args[1].get("arguments") or {} assert args.get(key) is True, f"Expected {key}=True, got {args.get(key)}" @then('r2plan-the parsed arguments should have "{key}" as bool false') def step_arg_bool_false(context: Any, key: str) -> None: call_args = context.r2_mock_svc.use_action.call_args args = call_args.kwargs.get("arguments") or call_args[1].get("arguments") or {} assert args.get(key) is False, f"Expected {key}=False, got {args.get(key)}" @then('r2plan-the parsed arguments should have "{key}" as string "{val}"') def step_arg_string(context: Any, key: str, val: str) -> None: call_args = context.r2_mock_svc.use_action.call_args args = call_args.kwargs.get("arguments") or call_args[1].get("arguments") or {} assert args.get(key) == val, f"Expected {key}='{val}', got {args.get(key)}" assert isinstance(args[key], str)