"""Step definitions for plan lifecycle CLI coverage.""" from __future__ import annotations import importlib from types import SimpleNamespace from unittest.mock import MagicMock, patch from behave import given, then, when from typer.testing import CliRunner from cleveragents.application.services.plan_lifecycle_service import ( ActionNotAvailableError, InvalidPhaseTransitionError, PlanLifecycleService, PlanNotReadyError, ) from cleveragents.cli.commands import plan as plan_module from cleveragents.cli.commands.plan import app as plan_app from cleveragents.core.exceptions import ( CleverAgentsError, NotFoundError, PlanError, ValidationError, ) from cleveragents.domain.models.core.action import ActionState from cleveragents.domain.models.core.plan import ( NamespacedName, Plan, PlanIdentity, PlanPhase, ProcessingState, ProjectLink, ) _ULIDS = [ "01ARZ3NDEKTSV4RRFFQ69G5FAV", "01ARZ3NDEKTSV4RRFFQ69G5FAW", "01ARZ3NDEKTSV4RRFFQ69G5FAX", "01ARZ3NDEKTSV4RRFFQ69G5FAY", "01ARZ3NDEKTSV4RRFFQ69G5FAZ", "01ARZ3NDEKTSV4RRFFQ69G5FB0", "01ARZ3NDEKTSV4RRFFQ69G5FB1", "01ARZ3NDEKTSV4RRFFQ69G5FB2", "01ARZ3NDEKTSV4RRFFQ69G5FB3", "01ARZ3NDEKTSV4RRFFQ69G5FB4", "01ARZ3NDEKTSV4RRFFQ69G5FB5", "01ARZ3NDEKTSV4RRFFQ69G5FB6", "01ARZ3NDEKTSV4RRFFQ69G5FB7", "01ARZ3NDEKTSV4RRFFQ69G5FB8", "01ARZ3NDEKTSV4RRFFQ69G5FB9", "01ARZ3NDEKTSV4RRFFQ69G5FBA", "01ARZ3NDEKTSV4RRFFQ69G5FBB", "01ARZ3NDEKTSV4RRFFQ69G5FBC", "01ARZ3NDEKTSV4RRFFQ69G5FBD", "01ARZ3NDEKTSV4RRFFQ69G5FBE", ] def _make_plan( *, plan_id: str, name: str, description: str = "Lifecycle plan", phase: PlanPhase = PlanPhase.STRATEGIZE, processing_state: ProcessingState = ProcessingState.COMPLETE, project_links: list[ProjectLink] | None = None, error_message: str | None = None, ) -> Plan: return Plan( identity=PlanIdentity(plan_id=plan_id), namespaced_name=NamespacedName.parse(name), action_name="local/test-action", description=description, definition_of_done=None, phase=phase, processing_state=processing_state, project_links=project_links or [], error_message=error_message, reusable=True, read_only=False, ) def _output(context) -> str: return getattr(context.result, "output", "") if hasattr(context, "result") else "" @given("a plan lifecycle CLI runner") def step_plan_lifecycle_cli_runner(context) -> None: context.runner = CliRunner() @given("a mocked lifecycle service for plan commands") def step_mocked_plan_lifecycle_service(context) -> None: context.lifecycle_service = MagicMock() patcher = patch( "cleveragents.cli.commands.plan._get_lifecycle_service", return_value=context.lifecycle_service, ) patcher.start() if not hasattr(context, "_cleanup_handlers"): context._cleanup_handlers = [] context._cleanup_handlers.append(patcher.stop) # Widen the Rich console so table columns do not wrap during tests console_patcher = patch.object(plan_module.console, "width", 200) console_patcher.start() context._cleanup_handlers.append(console_patcher.stop) @when("I call the plan lifecycle service helper") def step_call_lifecycle_service_helper(context) -> None: settings = MagicMock() service = PlanLifecycleService(settings=settings) container = SimpleNamespace(plan_lifecycle_service=MagicMock(return_value=service)) reloaded_plan_module = importlib.reload(plan_module) with patch( "cleveragents.application.container.get_container", return_value=container ): context.helper_settings = settings context.helper_service = reloaded_plan_module._get_lifecycle_service() @then("the lifecycle service helper should return a plan lifecycle service") def step_lifecycle_service_helper_returns(context) -> None: svc = context.helper_service assert isinstance(svc, PlanLifecycleService), ( f"Expected PlanLifecycleService, got {type(svc).__name__} " f"(module={type(svc).__module__})" ) assert svc.settings is context.helper_settings @when("I run plan lifecycle use with parsed arguments") def step_plan_use_with_parsed_arguments(context) -> None: action = SimpleNamespace(namespaced_name="local/code-coverage") context.lifecycle_service.get_action.side_effect = NotFoundError( resource_type="action", resource_id="missing", ) context.lifecycle_service.get_action_by_name.return_value = action plan = _make_plan( plan_id=_ULIDS[1], name="local/coverage-plan", description="x" * 240, error_message="Strategy failed", project_links=[ ProjectLink(project_name="proj-1"), ProjectLink(project_name="proj-2"), ], ) context.lifecycle_service.use_action.return_value = plan context.result = context.runner.invoke( plan_app, [ "use", "local/code-coverage", "--project", "proj-1", "--project", "proj-2", "--arg", "count=3", "--arg", "ratio=4.2", "--arg", "enabled=true", "--arg", "label=alpha", ], ) @when('I run plan lifecycle use with invalid argument "{arg_value}"') def step_plan_use_with_invalid_argument(context, arg_value: str) -> None: action = SimpleNamespace(namespaced_name="local/code-coverage") context.lifecycle_service.get_action.return_value = action context.result = context.runner.invoke( plan_app, [ "use", "local/code-coverage", "--project", "proj-1", "--arg", arg_value, ], ) @when('I run plan lifecycle use causing "{error_type}"') def step_plan_use_error(context, error_type: str) -> None: action = SimpleNamespace(namespaced_name="local/code-coverage") context.lifecycle_service.get_action.return_value = action if error_type == "action not available": context.lifecycle_service.use_action.side_effect = ActionNotAvailableError( str(action.namespaced_name), ActionState.ARCHIVED, ) elif error_type == "validation error": context.lifecycle_service.use_action.side_effect = ValidationError( "Invalid action" ) elif error_type == "general error": context.lifecycle_service.use_action.side_effect = CleverAgentsError( "Use failed" ) else: raise ValueError(f"Unhandled error type: {error_type}") context.result = context.runner.invoke( plan_app, [ "use", "local/code-coverage", "--project", "proj-1", ], ) @when("I run plan execute without a plan id with {count:d} complete plans") def step_plan_execute_without_plan_id(context, count: int) -> None: plans = [ _make_plan( plan_id=_ULIDS[idx], name=f"local/execute-plan-{idx}", phase=PlanPhase.STRATEGIZE, processing_state=ProcessingState.COMPLETE, ) for idx in range(count) ] context.lifecycle_service.list_plans.return_value = plans if count == 1: context.lifecycle_service.get_plan.return_value = plans[0] context.lifecycle_service.execute_plan.return_value = plans[0] context.execute_plans = plans # Mock the plan executor for successful execute paths mock_executor = MagicMock() executor_patcher = patch( "cleveragents.cli.commands.plan._get_plan_executor", return_value=mock_executor, ) executor_patcher.start() if not hasattr(context, "_cleanup_handlers"): context._cleanup_handlers = [] context._cleanup_handlers.append(executor_patcher.stop) context.result = context.runner.invoke(plan_app, ["execute"]) @when('I run plan execute for plan id "{plan_id}" causing "{error_type}"') def step_plan_execute_error(context, plan_id: str, error_type: str) -> None: # Provide a real plan so phase checks pass before the error is raised error_plan = _make_plan( plan_id=plan_id, name="local/error-plan", phase=PlanPhase.STRATEGIZE, processing_state=ProcessingState.COMPLETE, ) context.lifecycle_service.get_plan.return_value = error_plan if error_type == "invalid transition": context.lifecycle_service.execute_plan.side_effect = ( InvalidPhaseTransitionError( PlanPhase.STRATEGIZE, PlanPhase.EXECUTE, ) ) elif error_type == "plan not ready": context.lifecycle_service.execute_plan.side_effect = PlanNotReadyError( plan_id, PlanPhase.STRATEGIZE, ProcessingState.QUEUED, ) elif error_type == "general error": context.lifecycle_service.execute_plan.side_effect = CleverAgentsError( "Execute failed" ) else: raise ValueError(f"Unhandled error type: {error_type}") context.result = context.runner.invoke(plan_app, ["execute", plan_id]) @when("I run plan lifecycle apply without a plan id with {count:d} complete plans") def step_plan_apply_without_plan_id(context, count: int) -> None: plans = [ _make_plan( plan_id=_ULIDS[10 + idx], name=f"local/apply-plan-{idx}", phase=PlanPhase.EXECUTE, processing_state=ProcessingState.COMPLETE, ) for idx in range(count) ] context.lifecycle_service.list_plans.return_value = plans if count == 1: context.lifecycle_service.apply_plan.return_value = plans[0] context.apply_plans = plans context.result = context.runner.invoke(plan_app, ["apply", "--yes"]) @when('I run plan lifecycle apply for plan id "{plan_id}" causing "{error_type}"') def step_plan_apply_error(context, plan_id: str, error_type: str) -> None: if error_type == "invalid transition": context.lifecycle_service.apply_plan.side_effect = InvalidPhaseTransitionError( PlanPhase.EXECUTE, PlanPhase.APPLY, ) elif error_type == "plan not ready": context.lifecycle_service.apply_plan.side_effect = PlanNotReadyError( plan_id, PlanPhase.EXECUTE, ProcessingState.QUEUED, ) elif error_type == "general error": context.lifecycle_service.apply_plan.side_effect = CleverAgentsError( "Apply failed" ) else: raise ValueError(f"Unhandled error type: {error_type}") context.result = context.runner.invoke(plan_app, ["apply", "--yes", plan_id]) @when("I run plan status without a plan id and {count:d} plans exist") def step_plan_status_without_plan_id(context, count: int) -> None: if count == 0: context.lifecycle_service.list_plans.return_value = [] else: plans = [ _make_plan( plan_id=_ULIDS[5 + idx], name=f"local/status-plan-{idx}", phase=PlanPhase.EXECUTE if idx % 2 == 0 else PlanPhase.APPLY, processing_state=( ProcessingState.COMPLETE if idx % 2 == 0 else ProcessingState.QUEUED ), ) for idx in range(count) ] context.lifecycle_service.list_plans.return_value = plans context.status_plans = plans context.result = context.runner.invoke(plan_app, ["status"]) @when('I run plan status for legacy plan id "{plan_id}"') def step_plan_status_for_legacy_plan(context, plan_id: str) -> None: context.lifecycle_service.get_plan.return_value = "legacy-plan" context.result = context.runner.invoke(plan_app, ["status", plan_id]) @when("I run plan status causing a general error") def step_plan_status_error(context) -> None: context.lifecycle_service.list_plans.side_effect = CleverAgentsError( "Status failed" ) context.result = context.runner.invoke(plan_app, ["status"]) @when('I run plan lifecycle list with invalid phase "{phase}"') def step_plan_list_invalid_phase(context, phase: str) -> None: context.result = context.runner.invoke( plan_app, [ "list", "--phase", phase, ], ) @when("I run plan lifecycle list with no plans") def step_plan_list_no_plans(context) -> None: context.lifecycle_service.list_plans.return_value = [] context.result = context.runner.invoke(plan_app, ["list"]) @when('I run plan lifecycle list with plans and phase "{phase}"') def step_plan_list_with_plans(context, phase: str) -> None: plans = [ _make_plan( plan_id=_ULIDS[12], name="local/short-projects", phase=PlanPhase.EXECUTE, processing_state=ProcessingState.PROCESSING, project_links=[ProjectLink(project_name="proj-1")], ), _make_plan( plan_id=_ULIDS[13], name="local/multi-projects", phase=PlanPhase.APPLY, processing_state=ProcessingState.QUEUED, project_links=[ ProjectLink(project_name="proj-1"), ProjectLink(project_name="proj-2"), ProjectLink(project_name="proj-3"), ], ), ] context.lifecycle_service.list_plans.return_value = plans context.result = context.runner.invoke( plan_app, [ "list", "--phase", phase, "--project", "proj-1", ], ) @when("I run plan lifecycle list causing a general error") def step_plan_list_error(context) -> None: context.lifecycle_service.list_plans.side_effect = CleverAgentsError("List failed") context.result = context.runner.invoke(plan_app, ["list"]) @when('I run plan cancel for plan id "{plan_id}" with reason "{reason}"') def step_plan_cancel_with_reason(context, plan_id: str, reason: str) -> None: plan = _make_plan( plan_id=plan_id, name="local/cancel-plan", phase=PlanPhase.APPLY, processing_state=ProcessingState.PROCESSING, ) context.lifecycle_service.cancel_plan.return_value = plan context.result = context.runner.invoke( plan_app, [ "cancel", plan_id, "--reason", reason, ], ) @when('I run plan cancel for plan id "{plan_id}" without reason') def step_plan_cancel_without_reason(context, plan_id: str) -> None: plan = _make_plan( plan_id=plan_id, name="local/cancel-plan", phase=PlanPhase.APPLY, processing_state=ProcessingState.PROCESSING, ) context.lifecycle_service.cancel_plan.return_value = plan context.result = context.runner.invoke(plan_app, ["cancel", plan_id]) @when('I run plan cancel for plan id "{plan_id}" causing "{error_type}"') def step_plan_cancel_error(context, plan_id: str, error_type: str) -> None: if error_type == "plan error": context.lifecycle_service.cancel_plan.side_effect = PlanError("Cannot cancel") elif error_type == "general error": context.lifecycle_service.cancel_plan.side_effect = CleverAgentsError( "Cancel failed" ) else: raise ValueError(f"Unhandled error type: {error_type}") context.result = context.runner.invoke(plan_app, ["cancel", plan_id]) @then("the plan lifecycle command should succeed") def step_plan_lifecycle_command_succeeds(context) -> None: assert context.result.exit_code == 0, f"Output: {_output(context)}" @then("the plan lifecycle command should abort") def step_plan_lifecycle_command_aborts(context) -> None: assert context.result.exit_code != 0, f"Output: {_output(context)}" @then('the plan lifecycle output should contain "{text}"') def step_plan_lifecycle_output_contains(context, text: str) -> None: output = _output(context) assert text in output, f"Expected '{text}' in output: {output}" @then('the plan lifecycle output should not contain "{text}"') def step_plan_lifecycle_output_not_contains(context, text: str) -> None: output = _output(context) assert text not in output, f"Did not expect '{text}' in output: {output}" @then("the plan lifecycle use should pass parsed arguments") def step_plan_lifecycle_use_parsed_arguments(context) -> None: context.lifecycle_service.get_action_by_name.assert_called_once() kwargs = context.lifecycle_service.use_action.call_args.kwargs links = kwargs["project_links"] assert [link.project_name for link in links] == ["proj-1", "proj-2"] assert kwargs["arguments"] == { "count": 3, "ratio": 4.2, "enabled": True, "label": "alpha", } @then("the execute command should run the single ready plan") def step_execute_command_runs_single_plan(context) -> None: assert len(context.execute_plans) == 1 plan_id = context.execute_plans[0].identity.plan_id context.lifecycle_service.execute_plan.assert_called_once_with(plan_id) @then("the lifecycle apply command should run the single ready plan") def step_apply_runs_single_plan(context) -> None: assert len(context.apply_plans) == 1 plan_id = context.apply_plans[0].identity.plan_id context.lifecycle_service.apply_plan.assert_called_once_with(plan_id) # ------------------------------------------------------------------ # Regression: lifecycle-service sharing between CLI handler & executor # ------------------------------------------------------------------ @when("I run plan execute verifying lifecycle service sharing") def step_plan_execute_verify_sharing(context) -> None: """Run ``plan execute`` and capture the ``_get_plan_executor`` call. After Background has already mocked ``_get_lifecycle_service`` to return ``context.lifecycle_service``, we patch ``_get_plan_executor`` so we can later assert it was called with the **same** service instance. """ plan = _make_plan( plan_id=_ULIDS[0], name="local/shared-svc-plan", phase=PlanPhase.STRATEGIZE, processing_state=ProcessingState.COMPLETE, ) context.lifecycle_service.list_plans.return_value = [plan] context.lifecycle_service.get_plan.return_value = plan context.lifecycle_service.execute_plan.return_value = plan mock_executor = MagicMock() executor_patcher = patch( "cleveragents.cli.commands.plan._get_plan_executor", return_value=mock_executor, ) mock_get_executor = executor_patcher.start() if not hasattr(context, "_cleanup_handlers"): context._cleanup_handlers = [] context._cleanup_handlers.append(executor_patcher.stop) context._mock_get_plan_executor = mock_get_executor context.result = context.runner.invoke(plan_app, ["execute"]) @then("the plan executor should receive the same lifecycle service instance") def step_executor_shares_lifecycle_service(context) -> None: """Assert ``_get_plan_executor`` was called with the shared service.""" mock_fn = context._mock_get_plan_executor mock_fn.assert_called_once() call_kwargs = mock_fn.call_args.kwargs assert "lifecycle_service" in call_kwargs, ( "_get_plan_executor was not called with a lifecycle_service kwarg; " f"got kwargs: {call_kwargs}" ) assert call_kwargs["lifecycle_service"] is context.lifecycle_service, ( "Expected _get_plan_executor to receive the same lifecycle service " "instance that _get_lifecycle_service() returned, but it received " "a different object." )