"""Step definitions for plan_cli_coverage_boost_r2.feature. Targets remaining uncovered lines in cleveragents/cli/commands/plan.py: - Line 153: _plan_spec_dict execution_env_priority fallback to FALLBACK.value - Lines 785-786: apply command v3 lifecycle path (plan_id given) - Lines 876-940: _apply_with_id full body + error handlers - Line 1249: _get_plan_executor fallback when lifecycle_service is None - Lines 1838-1839: execute_plan plan not found - Lines 1898-1899: execute_plan PreflightRejection handler - Lines 1910-1911: execute_plan ValueError handler - Lines 2904, 2909, 2914: resume_plan rich output else "" branches - Line 3299: tree_decisions_cmd TABLE format show_superseded branch - Line 3331: tree_decisions_cmd TABLE format long chosen_option truncation """ from __future__ import annotations from datetime import datetime from unittest.mock import MagicMock, patch from behave import given, then, when from behave.runner import Context from typer.testing import CliRunner from cleveragents.application.services.plan_lifecycle_service import ( InvalidPhaseTransitionError, PlanNotReadyError, ) from cleveragents.application.services.plan_preflight_guardrail import ( PreflightCheckName, PreflightRejection, ) from cleveragents.cli.commands.plan import ( _plan_spec_dict, ) from cleveragents.cli.commands.plan import ( app as plan_app, ) from cleveragents.core.exceptions import CleverAgentsError from cleveragents.domain.models.core.decision import Decision, DecisionType from cleveragents.domain.models.core.plan import ( ExecutionEnvPriority, NamespacedName, Plan, PlanIdentity, PlanPhase, PlanTimestamps, ProcessingState, ) from cleveragents.domain.models.core.resume import ResumeSummary _ULID_A = "01ARZ3NDEKTSV4RRFFQ69G5FAV" _ULID_B = "01ARZ3NDEKTSV4RRFFQ69G5FBV" _PATCH_CONTAINER = "cleveragents.application.container.get_container" _PATCH_GET_LIFECYCLE = "cleveragents.cli.commands.plan._get_lifecycle_service" _PATCH_GET_EXECUTOR = "cleveragents.cli.commands.plan._get_plan_executor" _PATCH_GET_PROJECT = "cleveragents.cli.commands.plan._get_current_project" def _make_lifecycle_plan( *, plan_id: str = _ULID_A, name: str = "local/test-plan", phase: PlanPhase = PlanPhase.EXECUTE, processing_state: ProcessingState = ProcessingState.COMPLETE, read_only: bool = False, execution_environment: str | None = None, execution_env_priority: ExecutionEnvPriority | None = None, ) -> Plan: """Build a real Plan object for testing.""" plan = Plan( identity=PlanIdentity(plan_id=plan_id), namespaced_name=NamespacedName.parse(name), action_name="local/test-action", description="Coverage test plan", phase=phase, processing_state=processing_state, project_links=[], timestamps=PlanTimestamps( created_at=datetime(2025, 6, 15, 10, 0, 0), updated_at=datetime(2025, 6, 15, 11, 0, 0), ), reusable=True, read_only=read_only, ) if execution_environment is not None: plan.execution_environment = execution_environment if execution_env_priority is not None: plan.execution_env_priority = execution_env_priority return plan def _cleanup_patches(context: Context) -> None: """Stop all patches registered on context.""" for stop_fn in getattr(context, "_r2boost_cleanups", []): stop_fn() context._r2boost_cleanups = [] # ====================================================================== # Given — CLI runner # ====================================================================== @given("a r2boost CLI runner") def step_r2boost_cli_runner(context: Context) -> None: context.r2boost_runner = CliRunner() context.r2boost_result = None if not hasattr(context, "_r2boost_cleanups"): context._r2boost_cleanups = [] context.add_cleanup(lambda: _cleanup_patches(context)) # ====================================================================== # _plan_spec_dict: execution_env_priority fallback (line 153) # ====================================================================== @given('a v3 plan with execution_environment "{env}" and execution_env_priority None') def step_plan_with_exec_env_no_priority(context: Context, env: str) -> None: context.r2boost_plan = _make_lifecycle_plan( execution_environment=env, execution_env_priority=None, ) @when("I call _plan_spec_dict for r2 coverage") def step_call_plan_spec_dict_r2(context: Context) -> None: context.r2boost_spec_dict = _plan_spec_dict(context.r2boost_plan) @then('the r2 spec dict should contain "{key}" with value "{value}"') def step_r2_spec_dict_contains(context: Context, key: str, value: str) -> None: assert key in context.r2boost_spec_dict, ( f"Key '{key}' not in spec dict: {list(context.r2boost_spec_dict.keys())}" ) assert str(context.r2boost_spec_dict[key]) == value, ( f"Expected '{value}', got '{context.r2boost_spec_dict[key]}'" ) # ====================================================================== # apply command: v3 lifecycle happy path (lines 785-786, 876-925) # ====================================================================== def _make_apply_happy_service() -> MagicMock: """Create a lifecycle service mock that walks through the full apply flow.""" service = MagicMock() # Phase 1: pre_plan — Execute/complete pre_plan = _make_lifecycle_plan( phase=PlanPhase.EXECUTE, processing_state=ProcessingState.COMPLETE, ) # Phase 2: after apply_plan -> Apply/queued apply_queued = _make_lifecycle_plan( phase=PlanPhase.APPLY, processing_state=ProcessingState.QUEUED, ) # Phase 3: after start_apply -> Apply/processing apply_processing = _make_lifecycle_plan( phase=PlanPhase.APPLY, processing_state=ProcessingState.PROCESSING, ) # Phase 4: after complete_apply -> Apply/applied apply_applied = _make_lifecycle_plan( phase=PlanPhase.APPLY, processing_state=ProcessingState.APPLIED, ) # get_plan returns different results as apply progresses service.get_plan.side_effect = [ pre_plan, # First call: pre-flight check apply_queued, # After apply_plan: Apply/queued apply_processing, # After start_apply: Apply/processing apply_applied, # Final fetch after complete_apply ] service.apply_plan.return_value = None service.start_apply.return_value = None service.complete_apply.return_value = None service._complete_apply_if_queued.return_value = apply_applied return service @given("a mocked lifecycle service for r2boost apply happy path") def step_mock_apply_happy(context: Context) -> None: service = _make_apply_happy_service() p = patch(_PATCH_GET_LIFECYCLE, return_value=service) p.start() context._r2boost_cleanups.append(p.stop) context.r2boost_service = service @when("I invoke r2boost apply with plan_id in rich format") def step_invoke_apply_rich(context: Context) -> None: context.r2boost_result = context.r2boost_runner.invoke( plan_app, ["apply", "--yes", _ULID_A] ) @when("I invoke r2boost apply with plan_id and format json") def step_invoke_apply_json(context: Context) -> None: context.r2boost_result = context.r2boost_runner.invoke( plan_app, ["apply", "--yes", _ULID_A, "--format", "json"] ) # ====================================================================== # _apply_with_id: plan not found (lines 886-888) # ====================================================================== @given("a mocked lifecycle service for r2boost apply where plan is not found") def step_mock_apply_not_found(context: Context) -> None: service = MagicMock() service.get_plan.return_value = None p = patch(_PATCH_GET_LIFECYCLE, return_value=service) p.start() context._r2boost_cleanups.append(p.stop) # ====================================================================== # _apply_with_id: plan is read-only (lines 889-893) # ====================================================================== @given("a mocked lifecycle service for r2boost apply where plan is read-only") def step_mock_apply_read_only(context: Context) -> None: service = MagicMock() read_only_plan = _make_lifecycle_plan(read_only=True) service.get_plan.return_value = read_only_plan p = patch(_PATCH_GET_LIFECYCLE, return_value=service) p.start() context._r2boost_cleanups.append(p.stop) # ====================================================================== # _apply_with_id: error handlers (lines 927-940) # ====================================================================== @given( "a mocked lifecycle service for r2boost apply that raises InvalidPhaseTransitionError" ) def step_mock_apply_invalid_transition(context: Context) -> None: service = MagicMock() plan = _make_lifecycle_plan( phase=PlanPhase.EXECUTE, processing_state=ProcessingState.COMPLETE, ) service.get_plan.return_value = plan service.apply_plan.side_effect = InvalidPhaseTransitionError( from_phase=PlanPhase.EXECUTE, to_phase=PlanPhase.APPLY, ) p = patch(_PATCH_GET_LIFECYCLE, return_value=service) p.start() context._r2boost_cleanups.append(p.stop) @given("a mocked lifecycle service for r2boost apply that raises PlanNotReadyError") def step_mock_apply_not_ready(context: Context) -> None: service = MagicMock() plan = _make_lifecycle_plan( phase=PlanPhase.EXECUTE, processing_state=ProcessingState.COMPLETE, ) service.get_plan.return_value = plan service.apply_plan.side_effect = PlanNotReadyError( plan_id=_ULID_A, phase=PlanPhase.EXECUTE, state=ProcessingState.PROCESSING, ) p = patch(_PATCH_GET_LIFECYCLE, return_value=service) p.start() context._r2boost_cleanups.append(p.stop) @given("a mocked lifecycle service for r2boost apply that raises ValueError") def step_mock_apply_value_error(context: Context) -> None: service = MagicMock() plan = _make_lifecycle_plan( phase=PlanPhase.EXECUTE, processing_state=ProcessingState.COMPLETE, ) service.get_plan.return_value = plan service.apply_plan.side_effect = ValueError("missing API key for provider") p = patch(_PATCH_GET_LIFECYCLE, return_value=service) p.start() context._r2boost_cleanups.append(p.stop) @given("a mocked lifecycle service for r2boost apply that raises CleverAgentsError") def step_mock_apply_clever_agents_error(context: Context) -> None: service = MagicMock() plan = _make_lifecycle_plan( phase=PlanPhase.EXECUTE, processing_state=ProcessingState.COMPLETE, ) service.get_plan.return_value = plan service.apply_plan.side_effect = CleverAgentsError("general apply failure") p = patch(_PATCH_GET_LIFECYCLE, return_value=service) p.start() context._r2boost_cleanups.append(p.stop) # ====================================================================== # _get_plan_executor: lifecycle_service=None fallback (line 1249) # ====================================================================== @given("a mocked container for r2boost plan executor") def step_mock_container_for_executor(context: Context) -> None: if not hasattr(context, "_r2boost_cleanups"): context._r2boost_cleanups = [] mock_container = MagicMock() mock_container.provider_registry.return_value = MagicMock() mock_lifecycle_svc = MagicMock() mock_container.plan_lifecycle_service.return_value = mock_lifecycle_svc p_container = patch(_PATCH_CONTAINER, return_value=mock_container) p_container.start() context._r2boost_cleanups.append(p_container.stop) context.r2boost_mock_container = mock_container context.r2boost_mock_lifecycle_svc = mock_lifecycle_svc context.add_cleanup(lambda: _cleanup_patches(context)) @when("I call _get_plan_executor without lifecycle_service") def step_call_get_plan_executor_no_svc(context: Context) -> None: from cleveragents.cli.commands.plan import _get_plan_executor # lifecycle_service is None, so it should call _get_lifecycle_service() context.r2boost_executor = _get_plan_executor(lifecycle_service=None) @then("the r2boost plan executor should be returned") def step_verify_executor_returned(context: Context) -> None: assert context.r2boost_executor is not None # Verify that plan_lifecycle_service was called on the container context.r2boost_mock_container.plan_lifecycle_service.assert_called_once() # ====================================================================== # execute_plan: plan not found (lines 1838-1839) # ====================================================================== @given("a mocked lifecycle service for r2boost execute where plan vanishes") def step_mock_execute_plan_vanishes(context: Context) -> None: service = MagicMock() # First get_plan for read_only check returns a plan non_read_only_plan = _make_lifecycle_plan( phase=PlanPhase.STRATEGIZE, processing_state=ProcessingState.QUEUED, read_only=False, ) # Second get_plan (after phase check) returns None = plan vanished service.get_plan.side_effect = [non_read_only_plan, None] service.list_plans.return_value = [] mock_executor = MagicMock() p_svc = patch(_PATCH_GET_LIFECYCLE, return_value=service) p_exec = patch(_PATCH_GET_EXECUTOR, return_value=mock_executor) p_svc.start() p_exec.start() context._r2boost_cleanups.extend([p_svc.stop, p_exec.stop]) @when("I invoke r2boost execute with plan_id") def step_invoke_execute_with_plan_id(context: Context) -> None: context.r2boost_result = context.r2boost_runner.invoke( plan_app, ["execute", _ULID_A] ) # ====================================================================== # execute_plan: PreflightRejection handler (lines 1898-1899) # ====================================================================== @given("a mocked lifecycle service for r2boost execute that raises PreflightRejection") def step_mock_execute_preflight_rejection(context: Context) -> None: service = MagicMock() plan = _make_lifecycle_plan( phase=PlanPhase.STRATEGIZE, processing_state=ProcessingState.QUEUED, ) service.get_plan.return_value = plan service.list_plans.return_value = [] mock_executor = MagicMock() mock_executor.run_strategize.side_effect = PreflightRejection( PreflightCheckName.AUTOMATION_POLICY, "budget limit exceeded" ) p_svc = patch(_PATCH_GET_LIFECYCLE, return_value=service) p_exec = patch(_PATCH_GET_EXECUTOR, return_value=mock_executor) p_svc.start() p_exec.start() context._r2boost_cleanups.extend([p_svc.stop, p_exec.stop]) # ====================================================================== # execute_plan: ValueError handler (lines 1910-1911) # ====================================================================== @given("a mocked lifecycle service for r2boost execute that raises ValueError") def step_mock_execute_value_error(context: Context) -> None: service = MagicMock() plan = _make_lifecycle_plan( phase=PlanPhase.STRATEGIZE, processing_state=ProcessingState.QUEUED, ) service.get_plan.return_value = plan service.list_plans.return_value = [] mock_executor = MagicMock() mock_executor.run_strategize.side_effect = ValueError( "missing OPENAI_API_KEY in environment" ) p_svc = patch(_PATCH_GET_LIFECYCLE, return_value=service) p_exec = patch(_PATCH_GET_EXECUTOR, return_value=mock_executor) p_svc.start() p_exec.start() context._r2boost_cleanups.extend([p_svc.stop, p_exec.stop]) # ====================================================================== # resume_plan: rich output with no optional fields (lines 2904, 2909, 2914) # ====================================================================== @given("a mocked resume service for r2boost with no optional fields") def step_mock_resume_no_optional(context: Context) -> None: summary = ResumeSummary( plan_id=_ULID_A, phase="execute", processing_state="processing", last_completed_step=2, next_step_index=3, total_steps=5, decision_id=None, last_checkpoint_id=None, sandbox_ref=None, ) mock_resume_svc = MagicMock() mock_resume_svc.resume_plan.return_value = summary mock_lifecycle_svc = MagicMock() p_lifecycle = patch(_PATCH_GET_LIFECYCLE, return_value=mock_lifecycle_svc) p_resume = patch( "cleveragents.application.services.plan_resume_service.PlanResumeService", return_value=mock_resume_svc, ) p_lifecycle.start() p_resume.start() context._r2boost_cleanups.extend([p_lifecycle.stop, p_resume.stop]) @when("I invoke r2boost plan resume in rich format") def step_invoke_resume_rich(context: Context) -> None: context.r2boost_result = context.r2boost_runner.invoke( plan_app, ["resume", _ULID_A] ) # ====================================================================== # tree_decisions_cmd: TABLE format with show_superseded (line 3299) # ====================================================================== def _make_decision( *, decision_id: str, plan_id: str = _ULID_A, sequence_number: int = 0, question: str = "What approach to use?", chosen_option: str = "Option A", parent_decision_id: str | None = None, superseded_by: str | None = None, ) -> Decision: """Create a Decision for testing.""" return Decision( decision_id=decision_id, plan_id=plan_id, sequence_number=sequence_number, decision_type=DecisionType.STRATEGY_CHOICE, question=question, chosen_option=chosen_option, parent_decision_id=parent_decision_id, superseded_by=superseded_by, ) @given("a mocked decision service for r2boost tree with superseded decisions") def step_mock_tree_superseded(context: Context) -> None: mock_container = MagicMock() mock_decision_svc = MagicMock() decisions = [ _make_decision( decision_id="01ARZ3NDEKTSV4RRFFQ69G5FA1", sequence_number=0, question="Root question", chosen_option="Root choice", ), _make_decision( decision_id="01ARZ3NDEKTSV4RRFFQ69G5FA2", sequence_number=1, question="Superseded question", chosen_option="Old choice", parent_decision_id="01ARZ3NDEKTSV4RRFFQ69G5FA1", superseded_by="01ARZ3NDEKTSV4RRFFQ69G5FA3", ), _make_decision( decision_id="01ARZ3NDEKTSV4RRFFQ69G5FA3", sequence_number=2, question="Replacement question", chosen_option="New choice", parent_decision_id="01ARZ3NDEKTSV4RRFFQ69G5FA1", ), ] mock_decision_svc.list_decisions.return_value = decisions mock_container.decision_service.return_value = mock_decision_svc p = patch(_PATCH_CONTAINER, return_value=mock_container) p.start() context._r2boost_cleanups.append(p.stop) @when("I invoke r2boost tree in table format with show-superseded") def step_invoke_tree_table_superseded(context: Context) -> None: context.r2boost_result = context.r2boost_runner.invoke( plan_app, ["tree", _ULID_A, "--format", "table", "--show-superseded"], ) # ====================================================================== # tree_decisions_cmd: TABLE format long chosen_option (line 3331) # ====================================================================== @given("a mocked decision service for r2boost tree with long chosen option") def step_mock_tree_long_option(context: Context) -> None: mock_container = MagicMock() mock_decision_svc = MagicMock() # Create a decision with chosen_option longer than 30 chars long_option = "This is a very long chosen option that exceeds thirty characters for truncation testing" decisions = [ _make_decision( decision_id="01ARZ3NDEKTSV4RRFFQ69G5FA1", sequence_number=0, question="A question that is quite long and should be truncated at forty characters boundary", chosen_option=long_option, ), ] mock_decision_svc.list_decisions.return_value = decisions mock_container.decision_service.return_value = mock_decision_svc p = patch(_PATCH_CONTAINER, return_value=mock_container) p.start() context._r2boost_cleanups.append(p.stop) @when("I invoke r2boost tree in table format") def step_invoke_tree_table(context: Context) -> None: context.r2boost_result = context.r2boost_runner.invoke( plan_app, ["tree", _ULID_A, "--format", "table"] ) # ====================================================================== # Common then steps # ====================================================================== @then("the r2boost command should exit normally") def step_r2boost_exit_normal(context: Context) -> None: result = context.r2boost_result assert result is not None, "No result captured" assert result.exit_code == 0, ( f"Expected exit code 0, got {result.exit_code}.\n" f"Output: {result.output}\n" f"Exception: {result.exception}" ) @then("the r2boost command should abort") def step_r2boost_abort(context: Context) -> None: result = context.r2boost_result assert result is not None, "No result captured" assert result.exit_code != 0, ( f"Expected non-zero exit code, got {result.exit_code}.\nOutput: {result.output}" ) @then('the r2boost output should contain "{text}"') def step_r2boost_output_contains(context: Context, text: str) -> None: output = context.r2boost_result.output if context.r2boost_result else "" assert text in output, f"Expected '{text}' in output:\n{output}" @then('the r2boost output should not contain "{text}"') def step_r2boost_output_not_contains(context: Context, text: str) -> None: output = context.r2boost_result.output if context.r2boost_result else "" assert text not in output, f"Did not expect '{text}' in output:\n{output}"