"""Step definitions for copy-on-write sandbox coverage boost. Covers uncovered lines and branches in copy_on_write.py: - SandboxCreationError re-raise (lines 140-142) when original path is a file - get_path / commit / rollback with _sandbox_path = None - OSError handlers in commit (lines 252-254) and rollback (lines 311-313) - cleanup with _sandbox_path = None (branch 341:347) - cleanup when parent dir already removed (branch 344:347) - commit dst_dir creation branch (branch 242:244) - commit deleted-file-not-in-original branch (branch 249:247) All steps use the ``cowcb`` prefix to avoid collisions with existing steps. """ from __future__ import annotations import os import shutil import tempfile from behave import given, then, when from behave.runner import Context from cleveragents.infrastructure.sandbox.copy_on_write import CopyOnWriteSandbox from cleveragents.infrastructure.sandbox.protocol import ( SandboxCommitError, SandboxCreationError, SandboxRollbackError, SandboxStateError, SandboxStatus, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _init_cowcb_test_dir() -> str: """Create a temporary directory with some test files.""" test_dir = tempfile.mkdtemp(prefix="cowcb-test-dir-") with open(os.path.join(test_dir, "existing.txt"), "w") as f: f.write("original content") with open(os.path.join(test_dir, "to_delete.txt"), "w") as f: f.write("will be deleted") os.makedirs(os.path.join(test_dir, "data"), exist_ok=True) with open(os.path.join(test_dir, "data", "file.txt"), "w") as f: f.write("nested file") return test_dir def _cleanup_dir(path: str) -> None: """Remove a directory tree if it exists, ignoring errors.""" if path and os.path.exists(path): shutil.rmtree(path, ignore_errors=True) # --------------------------------------------------------------------------- # Given # --------------------------------------------------------------------------- @given("a cowcb temporary file instead of a directory") def step_cowcb_temp_file(ctx: Context) -> None: """Create a temporary regular file (not a directory) to use as original_path.""" fd, path = tempfile.mkstemp(prefix="cowcb-file-") os.write(fd, b"not a directory") os.close(fd) ctx.cowcb_file_path = path ctx.cowcb_sandbox = None ctx.cowcb_error = None ctx.cowcb_commit_result = None # Register cleanup if not hasattr(ctx, "_cleanup_handlers"): ctx._cleanup_handlers = [] ctx._cleanup_handlers.append( lambda: os.unlink(path) if os.path.exists(path) else None ) @given("a cowcb test directory is initialised") def step_cowcb_init_dir(ctx: Context) -> None: ctx.cowcb_test_dir = _init_cowcb_test_dir() ctx.cowcb_sandbox = None ctx.cowcb_error = None ctx.cowcb_commit_result = None if not hasattr(ctx, "_cleanup_handlers"): ctx._cleanup_handlers = [] ctx._cleanup_handlers.append(lambda: _cleanup_dir(ctx.cowcb_test_dir)) @given("a cowcb sandbox with status CREATED and sandbox_path None") def step_cowcb_created_none_path(ctx: Context) -> None: """Create a sandbox and force internal state: status=CREATED, _sandbox_path=None.""" test_dir = _init_cowcb_test_dir() ctx.cowcb_test_dir = test_dir sandbox = CopyOnWriteSandbox(resource_id="res-cowcb", original_path=test_dir) # Force internal state to simulate the edge case sandbox._status = SandboxStatus.CREATED sandbox._sandbox_path = None ctx.cowcb_sandbox = sandbox ctx.cowcb_error = None if not hasattr(ctx, "_cleanup_handlers"): ctx._cleanup_handlers = [] ctx._cleanup_handlers.append(lambda: _cleanup_dir(test_dir)) @given("a cowcb sandbox with status ACTIVE and sandbox_path None") def step_cowcb_active_none_path(ctx: Context) -> None: """Create a sandbox and force internal state: status=ACTIVE, _sandbox_path=None.""" test_dir = _init_cowcb_test_dir() ctx.cowcb_test_dir = test_dir sandbox = CopyOnWriteSandbox(resource_id="res-cowcb", original_path=test_dir) # Force internal state to simulate the edge case sandbox._status = SandboxStatus.ACTIVE sandbox._sandbox_path = None ctx.cowcb_sandbox = sandbox ctx.cowcb_error = None if not hasattr(ctx, "_cleanup_handlers"): ctx._cleanup_handlers = [] ctx._cleanup_handlers.append(lambda: _cleanup_dir(test_dir)) @given("a cowcb sandbox with status PENDING and sandbox_path None") def step_cowcb_pending_none_path(ctx: Context) -> None: """Create a sandbox in PENDING state with _sandbox_path=None (the default).""" test_dir = _init_cowcb_test_dir() ctx.cowcb_test_dir = test_dir sandbox = CopyOnWriteSandbox(resource_id="res-cowcb", original_path=test_dir) # PENDING is the default; _sandbox_path is already None ctx.cowcb_sandbox = sandbox ctx.cowcb_error = None if not hasattr(ctx, "_cleanup_handlers"): ctx._cleanup_handlers = [] ctx._cleanup_handlers.append(lambda: _cleanup_dir(test_dir)) @given('a cowcb sandbox is created for plan "{plan_id}"') def step_cowcb_create_sandbox(ctx: Context, plan_id: str) -> None: """Create a sandbox normally (CREATED state).""" ctx.cowcb_sandbox = CopyOnWriteSandbox( resource_id="res-cowcb", original_path=ctx.cowcb_test_dir, ) ctx.cowcb_sandbox.create(plan_id) # Register cleanup for the sandbox temp directory sandbox_path = ctx.cowcb_sandbox._sandbox_path if sandbox_path: parent = os.path.dirname(sandbox_path) ctx._cleanup_handlers.append(lambda: _cleanup_dir(parent)) @given('a cowcb sandbox is created and activated for plan "{plan_id}"') def step_cowcb_create_and_activate(ctx: Context, plan_id: str) -> None: """Create a sandbox and activate it by calling get_path.""" ctx.cowcb_sandbox = CopyOnWriteSandbox( resource_id="res-cowcb", original_path=ctx.cowcb_test_dir, ) ctx.cowcb_sandbox.create(plan_id) # Activate by resolving a path ctx.cowcb_sandbox.get_path("existing.txt") assert ctx.cowcb_sandbox.status == SandboxStatus.ACTIVE # Register cleanup for the sandbox temp directory sandbox_path = ctx.cowcb_sandbox._sandbox_path if sandbox_path: parent = os.path.dirname(sandbox_path) ctx._cleanup_handlers.append(lambda: _cleanup_dir(parent)) # --------------------------------------------------------------------------- # When # --------------------------------------------------------------------------- @when('a cowcb sandbox is created on that file for plan "{plan_id}"') def step_cowcb_create_on_file(ctx: Context, plan_id: str) -> None: """Attempt to create a sandbox with a file as original_path.""" ctx.cowcb_sandbox = CopyOnWriteSandbox( resource_id="res-cowcb", original_path=ctx.cowcb_file_path, ) try: ctx.cowcb_sandbox.create(plan_id) except SandboxCreationError as exc: ctx.cowcb_error = exc @when('cowcb get_path is called with "{path}"') def step_cowcb_get_path(ctx: Context, path: str) -> None: try: ctx.cowcb_sandbox.get_path(path) except SandboxStateError as exc: ctx.cowcb_error = exc @when("cowcb commit is called") def step_cowcb_commit(ctx: Context) -> None: try: ctx.cowcb_commit_result = ctx.cowcb_sandbox.commit() except (SandboxStateError, SandboxCommitError) as exc: ctx.cowcb_error = exc @when("cowcb rollback is called") def step_cowcb_rollback(ctx: Context) -> None: try: ctx.cowcb_sandbox.rollback() except (SandboxStateError, SandboxRollbackError) as exc: ctx.cowcb_error = exc @when("cowcb cleanup is called") def step_cowcb_cleanup(ctx: Context) -> None: ctx.cowcb_sandbox.cleanup() @when("the cowcb original directory is removed before commit") def step_cowcb_remove_original_before_commit(ctx: Context) -> None: """Modify a file in the sandbox and then patch shutil.copy2 to raise OSError so the commit sync loop fails when trying to copy the changed file back.""" from unittest.mock import patch as _mock_patch # First, modify a file in the sandbox so _compute_diff detects a change sandbox_path = ctx.cowcb_sandbox._sandbox_path if sandbox_path: modified = os.path.join(sandbox_path, "existing.txt") if os.path.exists(modified): with open(modified, "w") as f: f.write("modified content for OSError test") patcher = _mock_patch( "cleveragents.infrastructure.sandbox.copy_on_write.shutil.copy2", side_effect=OSError("mocked copy2 failure"), ) patcher.start() if not hasattr(ctx, "_cleanup_handlers"): ctx._cleanup_handlers = [] ctx._cleanup_handlers.append(patcher.stop) @when("the cowcb original directory is removed before rollback") def step_cowcb_remove_original_before_rollback(ctx: Context) -> None: """Remove the original directory so that shutil.copytree in rollback triggers OSError.""" shutil.rmtree(ctx.cowcb_test_dir) @when("the cowcb sandbox parent directory is removed") def step_cowcb_remove_sandbox_parent(ctx: Context) -> None: """Remove the sandbox's parent temp directory before cleanup.""" sandbox_path = ctx.cowcb_sandbox._sandbox_path if sandbox_path: parent = os.path.dirname(sandbox_path) if os.path.exists(parent): shutil.rmtree(parent, ignore_errors=True) @when('a cowcb file "{rel_path}" is created in the sandbox') def step_cowcb_create_file_in_sandbox(ctx: Context, rel_path: str) -> None: """Create a new file at a relative path inside the sandbox copy.""" sandbox_path = ctx.cowcb_sandbox._sandbox_path assert sandbox_path is not None file_path = os.path.join(sandbox_path, rel_path) os.makedirs(os.path.dirname(file_path), exist_ok=True) with open(file_path, "w") as f: f.write("coverage boost content") @when('the cowcb file "{filename}" is deleted from both sandbox and original') def step_cowcb_delete_from_both(ctx: Context, filename: str) -> None: """Delete a file from both sandbox and original so commit's delete loop encounters a file that doesn't exist in the original.""" sandbox_path = ctx.cowcb_sandbox._sandbox_path assert sandbox_path is not None # Delete from sandbox sandbox_file = os.path.join(sandbox_path, filename) if os.path.exists(sandbox_file): os.remove(sandbox_file) # Delete from original original_file = os.path.join(ctx.cowcb_test_dir, filename) if os.path.exists(original_file): os.remove(original_file) # --------------------------------------------------------------------------- # Then # --------------------------------------------------------------------------- @then("a cowcb SandboxCreationError should be raised") def step_cowcb_check_creation_error(ctx: Context) -> None: assert ctx.cowcb_error is not None, ( "Expected SandboxCreationError but none occurred" ) assert isinstance(ctx.cowcb_error, SandboxCreationError), ( f"Expected SandboxCreationError, got {type(ctx.cowcb_error).__name__}: {ctx.cowcb_error}" ) @then('a cowcb SandboxStateError should be raised with message "{msg}"') def step_cowcb_check_state_error_msg(ctx: Context, msg: str) -> None: assert ctx.cowcb_error is not None, "Expected SandboxStateError but none occurred" assert isinstance(ctx.cowcb_error, SandboxStateError), ( f"Expected SandboxStateError, got {type(ctx.cowcb_error).__name__}: {ctx.cowcb_error}" ) assert msg in str(ctx.cowcb_error), ( f"Expected '{msg}' in error message, got: {ctx.cowcb_error}" ) @then("a cowcb SandboxCommitError should be raised") def step_cowcb_check_commit_error(ctx: Context) -> None: assert ctx.cowcb_error is not None, "Expected SandboxCommitError but none occurred" assert isinstance(ctx.cowcb_error, SandboxCommitError), ( f"Expected SandboxCommitError, got {type(ctx.cowcb_error).__name__}: {ctx.cowcb_error}" ) @then("a cowcb SandboxRollbackError should be raised") def step_cowcb_check_rollback_error(ctx: Context) -> None: assert ctx.cowcb_error is not None, ( "Expected SandboxRollbackError but none occurred" ) assert isinstance(ctx.cowcb_error, SandboxRollbackError), ( f"Expected SandboxRollbackError, got {type(ctx.cowcb_error).__name__}: {ctx.cowcb_error}" ) @then('the cowcb sandbox status should be "{status}"') def step_cowcb_check_status(ctx: Context, status: str) -> None: expected = SandboxStatus(status) assert ctx.cowcb_sandbox.status == expected, ( f"Expected status {expected}, got {ctx.cowcb_sandbox.status}" ) @then("the cowcb commit should succeed") def step_cowcb_commit_success(ctx: Context) -> None: assert ctx.cowcb_error is None, f"Expected no error but got: {ctx.cowcb_error}" assert ctx.cowcb_commit_result is not None, "Expected a commit result" assert ctx.cowcb_commit_result.success is True, "Expected commit to succeed" @then('the cowcb file "{rel_path}" should exist in the original directory') def step_cowcb_file_in_original(ctx: Context, rel_path: str) -> None: file_path = os.path.join(ctx.cowcb_test_dir, rel_path) assert os.path.exists(file_path), ( f"File {rel_path} not found in original directory at {file_path}" )