"""Steps for sandbox_cleanup_conditional.feature.""" from __future__ import annotations from unittest.mock import MagicMock, patch from behave import given, then, when from typer.testing import CliRunner from cleveragents.cli.commands.plan import app as plan_app from cleveragents.domain.models.core.plan import PlanPhase, ProcessingState from cleveragents.infrastructure.sandbox.git_worktree import ( GitWorktreeSandbox, ) # --------------------------------------------------------------------------- # Fixture helper: mock sandbox objects # --------------------------------------------------------------------------- def _make_mock_sandbox_infos(count: int) -> list[object]: """Return a list of ``_SandboxInfo``-like objects with mocked cleanup. Since ``_SandboxInfo`` uses ``__slots__``, we cannot monkey-patch extra attributes. Instead we pass a MagicMock as ``sandbox_obj`` whose ``cleanup`` method we can later assert on via ``info.sandbox_obj.cleanup``. """ from cleveragents.cli.commands.plan import _SandboxInfo infos: list[object] = [] for i in range(count): sandbox_obj = MagicMock(spec=GitWorktreeSandbox) sandbox_obj.cleanup = MagicMock() sandbox_obj.sandbox_path = f"/tmp/sandbox-{i}" info = _SandboxInfo( sandbox_path=sandbox_obj.sandbox_path, sandbox_obj=sandbox_obj, resource_location=f"/tmp/resource-{i}", project_name=f"project-{i}", ) infos.append(info) return infos # --------------------------------------------------------------------------- # Context label dispatch helper # --------------------------------------------------------------------------- _LABEL_MAP = { "scco": "scco", "scef": "scef", "scef-er": "scef_er", "sccs": "sccs", } def _attr(context: object, label: str, suffix: str) -> object: """Get a context attribute by label and suffix, e.g. scco_runner.""" key = _LABEL_MAP.get(label) if key is None: raise ValueError(f"Unknown context_label: {label}") return getattr(context, f"{key}_{suffix}") # --------------------------------------------------------------------------- # scco - sandbox cleanup conditional on execute success # --------------------------------------------------------------------------- @given("a mocked lifecycle service with a strat-complete plan for scco") def step_mock_lifecycle_service_scco(context: object) -> None: plan = MagicMock() plan.phase = PlanPhase.STRATEGIZE plan.state = ProcessingState.COMPLETE plan.processing_state = ProcessingState.COMPLETE plan.identity.plan_id = "01TESTSCCPASS0000000000" plan.read_only = False plan.project_links = [] plan.automation_profile = None plan.timestamps = MagicMock() plan.timestamps.execute_started_at = None plan.estimation_result = None plan.decisions = [] plan.invariants = [] mock_service = MagicMock() # execute_plan transitions to EXECUTE/QUEUED so run_execute is called def fake_execute(pid: str) -> MagicMock: plan.phase = PlanPhase.EXECUTE plan.state = ProcessingState.QUEUED plan.processing_state = ProcessingState.QUEUED return plan mock_service.execute_plan.side_effect = fake_execute mock_service.get_plan.return_value = plan context.scco_runner = CliRunner() context.scco_service = mock_service context.scco_executor = MagicMock() context.scco_plan = plan @given("the executor completes successfully for scco") def step_executor_succeeds_scco(context: object) -> None: """No side-effect — ``run_execute`` does nothing → success.""" context.scco_executor.run_execute.side_effect = None @given("sandbox_infos contains two active sandboxes for scco") def step_two_sandboxes_scco(context: object) -> None: context.scco_sandbox_infos = _make_mock_sandbox_infos(2) context.scco_sandbox_patcher = patch( "cleveragents.cli.commands.plan._create_sandbox_for_plan", return_value=("/tmp/sandbox-parent", context.scco_sandbox_infos), ) context.scco_sandbox_patcher.start() if not hasattr(context, "_cleanup_handlers"): context._cleanup_handlers = [] context._cleanup_handlers.append(context.scco_sandbox_patcher.stop) context.scco_executor_patcher = patch( "cleveragents.cli.commands.plan._get_plan_executor", return_value=context.scco_executor, ) context.scco_executor_patcher.start() context._cleanup_handlers.append(context.scco_executor_patcher.stop) @given("a mocked lifecycle service with a strat-complete plan for sccs") def step_mock_lifecycle_service_sccs(context: object) -> None: plan = MagicMock() plan.phase = PlanPhase.STRATEGIZE plan.state = ProcessingState.COMPLETE plan.processing_state = ProcessingState.COMPLETE plan.identity.plan_id = "01TESTSCCSKIP000000000000" plan.read_only = False plan.project_links = [] plan.automation_profile = None plan.timestamps = MagicMock() plan.timestamps.execute_started_at = None plan.estimation_result = None plan.decisions = [] plan.invariants = [] mock_service = MagicMock() # execute_plan transitions to EXECUTE/QUEUED so run_execute is called def fake_execute(pid: str) -> MagicMock: plan.phase = PlanPhase.EXECUTE plan.state = ProcessingState.QUEUED plan.processing_state = ProcessingState.QUEUED return plan mock_service.execute_plan.side_effect = fake_execute mock_service.get_plan.return_value = plan context.sccs_runner = CliRunner() context.sccs_service = mock_service context.sccs_executor = MagicMock() context.sccs_plan = plan @given("the executor completes successfully for sccs") def step_executor_succeeds_sccs(context: object) -> None: context.sccs_executor.run_execute.side_effect = None @given("sandbox_infos is empty for sccs") def step_empty_sandboxes_sccs(context: object) -> None: context.sccs_sandbox_infos = [] context.sccs_sandbox_patcher = patch( "cleveragents.cli.commands.plan._create_sandbox_for_plan", return_value=(None, []), ) context.sccs_sandbox_patcher.start() if not hasattr(context, "_cleanup_handlers"): context._cleanup_handlers = [] context._cleanup_handlers.append(context.sccs_sandbox_patcher.stop) context.sccs_executor_patcher = patch( "cleveragents.cli.commands.plan._get_plan_executor", return_value=context.sccs_executor, ) context.sccs_executor_patcher.start() context._cleanup_handlers.append(context.sccs_executor_patcher.stop) # --------------------------------------------------------------------------- # scef - sandbox cleanup on execute failure # --------------------------------------------------------------------------- @given("a mocked lifecycle service with a strat-complete plan for scef") def step_mock_lifecycle_service_scef(context: object) -> None: plan = MagicMock() plan.phase = PlanPhase.STRATEGIZE plan.state = ProcessingState.COMPLETE plan.processing_state = ProcessingState.COMPLETE plan.identity.plan_id = "01TESTSCCFAIL0000000000" plan.read_only = False plan.project_links = [] plan.automation_profile = None plan.timestamps = MagicMock() plan.timestamps.execute_started_at = None plan.estimation_result = None plan.decisions = [] plan.invariants = [] mock_service = MagicMock() # execute_plan transitions to EXECUTE/QUEUED so run_execute is called def fake_execute(pid: str) -> MagicMock: plan.phase = PlanPhase.EXECUTE plan.state = ProcessingState.QUEUED plan.processing_state = ProcessingState.QUEUED return plan mock_service.execute_plan.side_effect = fake_execute mock_service.get_plan.return_value = plan context.scef_runner = CliRunner() context.scef_service = mock_service context.scef_executor = MagicMock() context.scef_plan = plan @given("the executor raises RuntimeError during run_execute for scef") def step_executor_raises_scef(context: object) -> None: context.scef_executor.run_execute.side_effect = RuntimeError( "Simulated execute failure" ) @given("sandbox_infos contains two active sandboxes for scef") def step_two_sandboxes_scef(context: object) -> None: context.scef_sandbox_infos = _make_mock_sandbox_infos(2) context.scef_sandbox_patcher = patch( "cleveragents.cli.commands.plan._create_sandbox_for_plan", return_value=("/tmp/sandbox-parent", context.scef_sandbox_infos), ) context.scef_sandbox_patcher.start() if not hasattr(context, "_cleanup_handlers"): context._cleanup_handlers = [] context._cleanup_handlers.append(context.scef_sandbox_patcher.stop) context.scef_executor_patcher = patch( "cleveragents.cli.commands.plan._get_plan_executor", return_value=context.scef_executor, ) context.scef_executor_patcher.start() context._cleanup_handlers.append(context.scef_executor_patcher.stop) # --------------------------------------------------------------------------- # scef-er - sandbox cleanup on error-recovery failure # --------------------------------------------------------------------------- @given("a mocked lifecycle service with an execute-errored plan for scef-er") def step_mock_execute_errored_scefer(context: object) -> None: plan = MagicMock() plan.phase = PlanPhase.EXECUTE plan.state = ProcessingState.ERRORED plan.processing_state = ProcessingState.ERRORED plan.identity.plan_id = "01TESTSCEFERR0000000000" plan.error_details = {"exception_type": "ValueError"} plan.read_only = False plan.project_links = [] plan.automation_profile = None plan.timestamps = MagicMock() plan.timestamps.execute_started_at = None mock_service = MagicMock() mock_service.get_plan.return_value = plan # revert_plan raises ValueError → _recover_errored_execute_plan # converts it to typer.Abort → caught by the outer except → cleanup mock_service.revert_plan.side_effect = ValueError("Recovery failed") context.scef_er_runner = CliRunner() context.scef_er_service = mock_service context.scef_er_executor = MagicMock() context.scef_er_plan = plan @given("the executor raises ValueError during error recovery for scef-er") def step_executor_raises_value_error(context: object) -> None: """The ValueError is already wired on service.revert_plan — nothing to do.""" @given("sandbox_infos contains two active sandboxes for scef-er") def step_two_sandboxes_scefer(context: object) -> None: context.scef_er_sandbox_infos = _make_mock_sandbox_infos(2) context.scef_er_sandbox_patcher = patch( "cleveragents.cli.commands.plan._create_sandbox_for_plan", return_value=( "/tmp/sandbox-parent", context.scef_er_sandbox_infos, ), ) context.scef_er_sandbox_patcher.start() if not hasattr(context, "_cleanup_handlers"): context._cleanup_handlers = [] context._cleanup_handlers.append(context.scef_er_sandbox_patcher.stop) context.scef_er_executor_patcher = patch( "cleveragents.cli.commands.plan._get_plan_executor", return_value=context.scef_er_executor, ) context.scef_er_executor_patcher.start() context._cleanup_handlers.append(context.scef_er_executor_patcher.stop) # --------------------------------------------------------------------------- # When steps # --------------------------------------------------------------------------- @when('I invoke plan execute for plan "{plan_id}" for {context_label}') def step_invoke_execute(context: object, plan_id: str, context_label: str) -> None: """Invoke ``plan execute`` with the appropriate mocked fixtures.""" from contextlib import ExitStack # Map context_label to the correct attribute prefix prefix_map = { "scco": "scco", "scef": "scef", "sccs": "sccs", "scef-er": "scef_er", } prefix = prefix_map.get(context_label) if prefix is None: raise ValueError(f"Unknown context_label: {context_label}") runner = getattr(context, f"{prefix}_runner") sandboxes = getattr(context, f"{prefix}_sandbox_infos") executor = getattr(context, f"{prefix}_executor") svc = getattr(context, f"{prefix}_service") with ExitStack() as stack: stack.enter_context( patch( "cleveragents.cli.commands.plan._get_lifecycle_service", return_value=svc, ) ) stack.enter_context( patch( "cleveragents.cli.commands.plan._create_sandbox_for_plan", return_value=("/tmp/sandbox-parent", sandboxes), ) ) stack.enter_context( patch( "cleveragents.cli.commands.plan._get_plan_executor", return_value=executor, ) ) stack.enter_context( patch( "cleveragents.cli.commands.plan._validate_plan_ulid", ) ) stack.enter_context( patch( "cleveragents.cli.commands.plan._route_sandbox_files_to_worktrees", ) ) stack.enter_context( patch( "cleveragents.cli.commands.plan._commit_worktree_changes", ) ) stack.enter_context( patch( "cleveragents.cli.commands.plan._notify_facade", ) ) # For error-recovery scenarios the real _recover_errored_execute_plan # must run so the error propagates. For all others we stub it out. if context_label != "scef-er": stack.enter_context( patch( "cleveragents.cli.commands.plan._recover_errored_execute_plan", ) ) context.sandbox_result = runner.invoke(plan_app, ["execute", plan_id]) context.sandbox_output = context.sandbox_result.output # --------------------------------------------------------------------------- # Then steps # --------------------------------------------------------------------------- def _get_sandbox_infos(context: object, context_label: str) -> list[object]: """Return the sandbox_infos list for a given context label.""" prefix_map = { "scco": "scco", "scef": "scef", "sccs": "sccs", "scef-er": "scef_er", } prefix = prefix_map.get(context_label) if prefix is None: raise AssertionError(f"Unknown context_label: {context_label}") return getattr(context, f"{prefix}_sandbox_infos") @then("sandbox cleanup should have been called for {context_label}") def step_cleanup_called(context: object, context_label: str) -> None: """Verify every sandbox_obj in the infos called cleanup().""" sandboxes = _get_sandbox_infos(context, context_label) for info in sandboxes: assert info.sandbox_obj.cleanup.called, ( f"Expected sandbox_obj.cleanup() to be called, " f"but it was not for sandbox {info.sandbox_path}" ) @then("sandbox cleanup should NOT have been called for {context_label}") def step_cleanup_not_called(context: object, context_label: str) -> None: """Verify no sandbox_obj cleanup() was called.""" sandboxes = _get_sandbox_infos(context, context_label) for info in sandboxes: assert not info.sandbox_obj.cleanup.called, ( f"Expected sandbox_obj.cleanup() NOT to be called, " f"but it was called for sandbox {info.sandbox_path}" ) @then("the CLI exit code should be zero for {context_label}") def step_check_success(context: object, context_label: str) -> None: assert context.sandbox_result.exit_code == 0, ( f"Expected zero but got {context.sandbox_result.exit_code}\n" f"Output: {context.sandbox_output[:500]}" ) @then("the CLI exit code should be non-zero for {context_label}") def step_check_failure(context: object, context_label: str) -> None: assert context.sandbox_result.exit_code != 0, ( f"Expected non-zero but got {context.sandbox_result.exit_code}" )