"""Step definitions for checkpoint_service_coverage_boost.feature. These steps target specific uncovered lines in checkpoint_service.py: - Lines 484-487: subprocess.CalledProcessError handler in _run_git - Lines 488-491: subprocess.TimeoutExpired handler in _run_git - Line 417: ResourceNotFoundError when lifecycle service returns None plan """ import os import subprocess import tempfile from unittest.mock import MagicMock, patch from behave import given, then, when from ulid import ULID from cleveragents.application.services.checkpoint_service import CheckpointService from cleveragents.core.exceptions import BusinessRuleViolation, ResourceNotFoundError # Generate a stable ULID for the rollback test plan _ROLLBACK_PLAN_ID = str(ULID()) # --------------------------------------------------------------------------- # Background # --------------------------------------------------------------------------- @given("the checkpoint service module is imported for coverage boost") def step_checkpoint_module_imported(context): """Ensure the checkpoint service module is importable.""" assert CheckpointService is not None # --------------------------------------------------------------------------- # Shared: create a checkpoint service with in-memory storage # --------------------------------------------------------------------------- @given("a checkpoint service with in-memory storage for coverage boost") def step_create_in_memory_service(context): """Create a CheckpointService with no repository (in-memory mode).""" context.cp_service = CheckpointService() # --------------------------------------------------------------------------- # Scenario: Git CalledProcessError → BusinessRuleViolation (lines 484-487) # --------------------------------------------------------------------------- @given("subprocess.run is mocked to raise CalledProcessError") def step_mock_subprocess_called_process_error(context): """Prepare a mock that raises CalledProcessError.""" context.subprocess_side_effect = subprocess.CalledProcessError( returncode=128, cmd=["git", "status"], stderr="fatal: not a git repository", ) @when('I invoke _run_git with args "{git_args}"') def step_invoke_run_git(context, git_args): """Call _run_git with a mocked subprocess.run.""" args = git_args.split() context.run_git_error = None with patch("subprocess.run", side_effect=context.subprocess_side_effect): try: context.cp_service._run_git(args, cwd="/tmp/fake-sandbox") except (BusinessRuleViolation, Exception) as exc: context.run_git_error = exc @then('a BusinessRuleViolation should be raised mentioning "{fragment}"') def step_verify_business_rule_violation(context, fragment): """Verify a BusinessRuleViolation was raised containing the fragment.""" error = getattr(context, "run_git_error", None) or getattr( context, "rollback_error", None ) assert error is not None, "Expected an exception but none was raised" assert isinstance(error, BusinessRuleViolation), ( f"Expected BusinessRuleViolation, got {type(error).__name__}: {error}" ) assert fragment in str(error), ( f"Expected '{fragment}' in error message, got: {error}" ) @then('the error message should contain the git command "{cmd}"') def step_verify_error_contains_git_command(context, cmd): """Verify the error message includes the git subcommand.""" assert cmd in str(context.run_git_error), ( f"Expected '{cmd}' in error message, got: {context.run_git_error}" ) @then("the error message should include the stderr output") def step_verify_error_contains_stderr(context): """Verify the CalledProcessError handler includes stderr in the message.""" msg = str(context.run_git_error) assert "fatal:" in msg or "not a git repository" in msg, ( f"Expected stderr content in error message, got: {msg}" ) # --------------------------------------------------------------------------- # Scenario: Git TimeoutExpired → BusinessRuleViolation (lines 488-491) # --------------------------------------------------------------------------- @given("subprocess.run is mocked to raise TimeoutExpired") def step_mock_subprocess_timeout(context): """Prepare a mock that raises TimeoutExpired.""" context.subprocess_side_effect = subprocess.TimeoutExpired( cmd=["git", "log"], timeout=60, ) @then('the timeout error message should contain the git command "{cmd}"') def step_verify_timeout_error_contains_git_command(context, cmd): """Verify the timeout error message includes the git subcommand.""" assert context.run_git_error is not None, ( "Expected an exception but none was raised" ) assert isinstance(context.run_git_error, BusinessRuleViolation), ( f"Expected BusinessRuleViolation, got {type(context.run_git_error).__name__}" ) assert "timed out" in str(context.run_git_error), ( f"Expected 'timed out' in error message, got: {context.run_git_error}" ) assert cmd in str(context.run_git_error), ( f"Expected '{cmd}' in error message, got: {context.run_git_error}" ) # --------------------------------------------------------------------------- # Scenario: CalledProcessError during rollback (lines 484-487 via rollback) # --------------------------------------------------------------------------- @given("a sandbox directory exists at a temporary path") def step_create_temp_sandbox(context): """Create a temporary directory with a .git subdirectory to act as sandbox.""" context.sandbox_tmpdir = tempfile.mkdtemp(prefix="sandbox_test_") os.makedirs(os.path.join(context.sandbox_tmpdir, ".git"), exist_ok=True) context.rollback_plan_id = _ROLLBACK_PLAN_ID context.cp_service.register_sandbox( context.rollback_plan_id, context.sandbox_tmpdir ) def cleanup(): import shutil shutil.rmtree(context.sandbox_tmpdir, ignore_errors=True) context.add_cleanup(cleanup) @given("a checkpoint was created for the rollback test plan") def step_create_checkpoint_for_rollback(context): """Create a checkpoint for the rollback test plan.""" context.rollback_checkpoint = context.cp_service.create_checkpoint( plan_id=context.rollback_plan_id, sandbox_ref="abc123deadbeef", reason="test checkpoint for rollback", ) @given("subprocess.run is mocked to raise CalledProcessError for git diff") def step_mock_called_process_error_for_diff(context): """Mock subprocess.run to raise CalledProcessError.""" context.subprocess_patch = patch( "subprocess.run", side_effect=subprocess.CalledProcessError( returncode=1, cmd=["git", "diff", "--name-only"], stderr="fatal: bad revision", ), ) context.subprocess_mock = context.subprocess_patch.start() context.add_cleanup(context.subprocess_patch.stop) @when("I attempt to rollback to the checkpoint") def step_attempt_rollback(context): """Attempt rollback and capture any error.""" context.rollback_error = None context.run_git_error = None try: context.cp_service.rollback_to_checkpoint( plan_id=context.rollback_plan_id, checkpoint_id=context.rollback_checkpoint.checkpoint_id, ) except Exception as exc: context.rollback_error = exc context.run_git_error = exc # --------------------------------------------------------------------------- # Scenario: TimeoutExpired during rollback (lines 488-491 via rollback) # --------------------------------------------------------------------------- @given("subprocess.run is mocked to raise TimeoutExpired for git reset") def step_mock_timeout_for_reset(context): """Mock subprocess.run to raise TimeoutExpired.""" context.subprocess_patch = patch( "subprocess.run", side_effect=subprocess.TimeoutExpired( cmd=["git", "reset", "--hard"], timeout=60, ), ) context.subprocess_mock = context.subprocess_patch.start() context.add_cleanup(context.subprocess_patch.stop) # --------------------------------------------------------------------------- # Scenario: _resolve_sandbox_path with lifecycle service returning None # (line 417) # --------------------------------------------------------------------------- @given("a checkpoint service with a lifecycle service that returns None for any plan") def step_create_service_with_null_lifecycle(context): """Create a CheckpointService with a mocked lifecycle service that returns None.""" mock_lifecycle = MagicMock() mock_lifecycle.get_plan.return_value = None context.cp_service = CheckpointService(plan_lifecycle_service=mock_lifecycle) @when('I attempt to resolve the sandbox path for plan "{plan_id}"') def step_attempt_resolve_sandbox(context, plan_id): """Attempt to call _resolve_sandbox_path and capture the error.""" context.resolve_error = None try: context.cp_service._resolve_sandbox_path(plan_id) except Exception as exc: context.resolve_error = exc @then('a ResourceNotFoundError should be raised for resource type "{resource_type}"') def step_verify_resource_not_found(context, resource_type): """Verify a ResourceNotFoundError was raised with the expected resource type.""" assert context.resolve_error is not None, ( "Expected an exception but none was raised" ) assert isinstance(context.resolve_error, ResourceNotFoundError), ( f"Expected ResourceNotFoundError, got {type(context.resolve_error).__name__}: " f"{context.resolve_error}" ) assert context.resolve_error.resource_type == resource_type, ( f"Expected resource_type='{resource_type}', " f"got '{context.resolve_error.resource_type}'" ) @then('the ResourceNotFoundError should reference plan id "{plan_id}"') def step_verify_resource_not_found_plan_id(context, plan_id): """Verify the error references the correct plan ID.""" assert context.resolve_error.resource_id == plan_id, ( f"Expected resource_id='{plan_id}', got '{context.resolve_error.resource_id}'" )