forked from HAL9000/cleveragents-core
khyari hamza
cc45b5ff2c
Implement GitWorktreeSandbox (worktree creation, commit+merge, rollback, cleanup, branch sanitization, git timeouts) and CopyOnWriteSandbox (directory copy, diff-based commit, rollback, cleanup). Update SandboxFactory to wire new strategies, remove stale overlay/ versioning constants, add snapshot placeholder, align resource type mapping to spec types (git-checkout, fs-mount, fs-directory, fs-file). Add 40 Behave scenarios (19 git_worktree + 21 copy_on_write) with gwt/cow prefixed steps. Update existing sandbox_factory_coverage feature (21 scenarios) to reflect new factory behavior. All quality gates pass: ruff, pyright, 2325 scenarios, 97% coverage. TASK-006 / Stage B4
392 lines
13 KiB
Python
392 lines
13 KiB
Python
"""Step definitions for copy-on-write sandbox feature.
|
|
|
|
All steps use the ``cow`` prefix to avoid collisions with the 128+ existing
|
|
step files loaded globally by Behave.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
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 (
|
|
SandboxCreationError,
|
|
SandboxStateError,
|
|
SandboxStatus,
|
|
)
|
|
|
|
|
|
def _init_test_dir(ctx: Context) -> str:
|
|
"""Create a temporary directory with some test files."""
|
|
test_dir = tempfile.mkdtemp(prefix="cow-test-dir-")
|
|
# Create some initial files
|
|
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
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background / Given
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a cow test directory is initialised")
|
|
def step_cow_init_dir(ctx: Context) -> None:
|
|
ctx.cow_test_dir = _init_test_dir(ctx)
|
|
ctx.cow_sandbox = None
|
|
ctx.cow_error = None
|
|
ctx.cow_commit_result = None
|
|
ctx.cow_resolved_path = None
|
|
|
|
|
|
@given("a cow non-existent directory")
|
|
def step_cow_nonexistent_dir(ctx: Context) -> None:
|
|
ctx.cow_nonexistent_dir = os.path.join(
|
|
tempfile.gettempdir(), "cow-nonexistent-" + str(os.getpid())
|
|
)
|
|
# Make sure it doesn't exist
|
|
if os.path.exists(ctx.cow_nonexistent_dir):
|
|
os.rmdir(ctx.cow_nonexistent_dir)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When - instantiation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("a cow sandbox is instantiated")
|
|
def step_cow_instantiate(ctx: Context) -> None:
|
|
ctx.cow_sandbox = CopyOnWriteSandbox(
|
|
resource_id="res-001",
|
|
original_path=ctx.cow_test_dir,
|
|
)
|
|
|
|
|
|
@when("a cow sandbox is prepared with empty resource_id")
|
|
def step_cow_empty_resource_id(ctx: Context) -> None:
|
|
try:
|
|
CopyOnWriteSandbox(resource_id="", original_path=ctx.cow_test_dir)
|
|
except ValueError as exc:
|
|
ctx.cow_error = exc
|
|
|
|
|
|
@when("a cow sandbox is prepared with empty original_path")
|
|
def step_cow_empty_original_path(ctx: Context) -> None:
|
|
try:
|
|
CopyOnWriteSandbox(resource_id="res-001", original_path="")
|
|
except ValueError as exc:
|
|
ctx.cow_error = exc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When - create
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('a cow sandbox is created for plan "{plan_id}"')
|
|
def step_cow_create(ctx: Context, plan_id: str) -> None:
|
|
ctx.cow_sandbox = CopyOnWriteSandbox(
|
|
resource_id="res-001",
|
|
original_path=ctx.cow_test_dir,
|
|
)
|
|
ctx.cow_sandbox.create(plan_id)
|
|
|
|
|
|
@when("a cow sandbox is created with empty plan_id")
|
|
def step_cow_create_empty_plan(ctx: Context) -> None:
|
|
ctx.cow_sandbox = CopyOnWriteSandbox(
|
|
resource_id="res-001",
|
|
original_path=ctx.cow_test_dir,
|
|
)
|
|
try:
|
|
ctx.cow_sandbox.create("")
|
|
except ValueError as exc:
|
|
ctx.cow_error = exc
|
|
|
|
|
|
@when('a cow sandbox is created on the non-existent directory for plan "{plan_id}"')
|
|
def step_cow_create_nonexistent(ctx: Context, plan_id: str) -> None:
|
|
ctx.cow_sandbox = CopyOnWriteSandbox(
|
|
resource_id="res-001",
|
|
original_path=ctx.cow_nonexistent_dir,
|
|
)
|
|
try:
|
|
ctx.cow_sandbox.create(plan_id)
|
|
except SandboxCreationError as exc:
|
|
ctx.cow_error = exc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When - path resolution
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('the cow path "{path}" is resolved')
|
|
def step_cow_resolve_path(ctx: Context, path: str) -> None:
|
|
try:
|
|
ctx.cow_resolved_path = ctx.cow_sandbox.get_path(path)
|
|
except (ValueError, SandboxStateError) as exc:
|
|
ctx.cow_error = exc
|
|
|
|
|
|
@when('the cow path "{path}" is resolved on a cleaned-up sandbox')
|
|
def step_cow_resolve_path_cleaned(ctx: Context, path: str) -> None:
|
|
try:
|
|
ctx.cow_sandbox.get_path(path)
|
|
except SandboxStateError as exc:
|
|
ctx.cow_error = exc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When - file operations in sandbox copy
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('a cow file "{filename}" is created in the sandbox with content "{content}"')
|
|
def step_cow_create_file(ctx: Context, filename: str, content: str) -> None:
|
|
sandbox_path = ctx.cow_sandbox.context.sandbox_path
|
|
file_path = os.path.join(sandbox_path, filename)
|
|
os.makedirs(os.path.dirname(file_path), exist_ok=True)
|
|
with open(file_path, "w") as f:
|
|
f.write(content)
|
|
|
|
|
|
@when(
|
|
'the cow existing file "{filename}" is modified in the sandbox with content "{content}"'
|
|
)
|
|
def step_cow_modify_file(ctx: Context, filename: str, content: str) -> None:
|
|
sandbox_path = ctx.cow_sandbox.context.sandbox_path
|
|
file_path = os.path.join(sandbox_path, filename)
|
|
with open(file_path, "w") as f:
|
|
f.write(content)
|
|
|
|
|
|
@when('the cow existing file "{filename}" is deleted from the sandbox')
|
|
def step_cow_delete_file(ctx: Context, filename: str) -> None:
|
|
sandbox_path = ctx.cow_sandbox.context.sandbox_path
|
|
file_path = os.path.join(sandbox_path, filename)
|
|
os.remove(file_path)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When - commit / rollback / cleanup
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("the cow sandbox is committed")
|
|
def step_cow_commit(ctx: Context) -> None:
|
|
try:
|
|
ctx.cow_commit_result = ctx.cow_sandbox.commit()
|
|
except (SandboxStateError, Exception) as exc:
|
|
ctx.cow_error = exc
|
|
|
|
|
|
@when("the cow sandbox commit is attempted on cleaned-up sandbox")
|
|
def step_cow_commit_cleaned(ctx: Context) -> None:
|
|
try:
|
|
ctx.cow_sandbox.commit()
|
|
except SandboxStateError as exc:
|
|
ctx.cow_error = exc
|
|
|
|
|
|
@when("the cow sandbox is rolled back")
|
|
def step_cow_rollback(ctx: Context) -> None:
|
|
try:
|
|
ctx.cow_sandbox.rollback()
|
|
except (SandboxStateError, Exception) as exc:
|
|
ctx.cow_error = exc
|
|
|
|
|
|
@when("the cow sandbox rollback is attempted on created sandbox")
|
|
def step_cow_rollback_created(ctx: Context) -> None:
|
|
try:
|
|
ctx.cow_sandbox.rollback()
|
|
except SandboxStateError as exc:
|
|
ctx.cow_error = exc
|
|
|
|
|
|
@when("the cow sandbox is cleaned up")
|
|
def step_cow_cleanup(ctx: Context) -> None:
|
|
ctx.cow_sandbox.cleanup()
|
|
|
|
|
|
@when("the cow sandbox is cleaned up again")
|
|
def step_cow_cleanup_again(ctx: Context) -> None:
|
|
ctx.cow_sandbox.cleanup()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then - status / state
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then('the cow sandbox should be in the "{status}" state')
|
|
def step_cow_check_status(ctx: Context, status: str) -> None:
|
|
expected = SandboxStatus(status)
|
|
assert ctx.cow_sandbox.status == expected, (
|
|
f"Expected status {expected}, got {ctx.cow_sandbox.status}"
|
|
)
|
|
|
|
|
|
@then('the cow sandbox context should reference plan "{plan_id}"')
|
|
def step_cow_check_plan(ctx: Context, plan_id: str) -> None:
|
|
assert ctx.cow_sandbox.context is not None
|
|
assert ctx.cow_sandbox.context.plan_id == plan_id
|
|
|
|
|
|
@then('the cow sandbox context should have strategy metadata "{strategy}"')
|
|
def step_cow_check_strategy(ctx: Context, strategy: str) -> None:
|
|
assert ctx.cow_sandbox.context is not None
|
|
assert ctx.cow_sandbox.context.metadata.get("strategy") == strategy
|
|
|
|
|
|
@then("the cow sandbox path should exist")
|
|
def step_cow_path_exists(ctx: Context) -> None:
|
|
assert ctx.cow_sandbox.context is not None
|
|
assert os.path.isdir(ctx.cow_sandbox.context.sandbox_path)
|
|
|
|
|
|
@then("the cow sandbox path should not exist")
|
|
def step_cow_path_not_exists(ctx: Context) -> None:
|
|
if ctx.cow_sandbox.context is not None:
|
|
assert not os.path.exists(ctx.cow_sandbox.context.sandbox_path)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then - errors
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then('a cow ValueError should be raised with message "{msg}"')
|
|
def step_cow_valueerror(ctx: Context, msg: str) -> None:
|
|
assert ctx.cow_error is not None, "Expected an error but none occurred"
|
|
assert isinstance(ctx.cow_error, ValueError), (
|
|
f"Expected ValueError, got {type(ctx.cow_error).__name__}"
|
|
)
|
|
assert msg in str(ctx.cow_error), (
|
|
f"Expected '{msg}' in error message, got: {ctx.cow_error}"
|
|
)
|
|
|
|
|
|
@then("a cow SandboxCreationError should be raised")
|
|
def step_cow_creation_error(ctx: Context) -> None:
|
|
assert ctx.cow_error is not None
|
|
assert isinstance(ctx.cow_error, SandboxCreationError)
|
|
|
|
|
|
@then("a cow SandboxStateError should be raised")
|
|
def step_cow_state_error(ctx: Context) -> None:
|
|
assert ctx.cow_error is not None
|
|
assert isinstance(ctx.cow_error, SandboxStateError)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then - path resolution
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the cow resolved path should be inside the sandbox copy")
|
|
def step_cow_path_in_sandbox(ctx: Context) -> None:
|
|
assert ctx.cow_resolved_path is not None
|
|
sandbox_path = ctx.cow_sandbox.context.sandbox_path
|
|
assert ctx.cow_resolved_path.startswith(sandbox_path)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then - commit results
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the cow commit result should indicate success")
|
|
def step_cow_commit_success(ctx: Context) -> None:
|
|
assert ctx.cow_commit_result is not None
|
|
assert ctx.cow_commit_result.success is True
|
|
|
|
|
|
@then("the cow commit result should have {count:d} changed files")
|
|
def step_cow_commit_changed(ctx: Context, count: int) -> None:
|
|
assert ctx.cow_commit_result is not None
|
|
assert len(ctx.cow_commit_result.changed_files) == count, (
|
|
f"Expected {count} changed, got {len(ctx.cow_commit_result.changed_files)}"
|
|
)
|
|
|
|
|
|
@then("the cow commit result should have {count:d} added files")
|
|
def step_cow_commit_added(ctx: Context, count: int) -> None:
|
|
assert ctx.cow_commit_result is not None
|
|
assert len(ctx.cow_commit_result.added_files) == count, (
|
|
f"Expected {count} added, got {len(ctx.cow_commit_result.added_files)}"
|
|
)
|
|
|
|
|
|
@then("the cow commit result should have {count:d} deleted files")
|
|
def step_cow_commit_deleted(ctx: Context, count: int) -> None:
|
|
assert ctx.cow_commit_result is not None
|
|
assert len(ctx.cow_commit_result.deleted_files) == count, (
|
|
f"Expected {count} deleted, got {len(ctx.cow_commit_result.deleted_files)}"
|
|
)
|
|
|
|
|
|
@then(
|
|
'the cow file "{filename}" should exist in the original directory with content "{content}"'
|
|
)
|
|
def step_cow_file_in_original_with_content(
|
|
ctx: Context, filename: str, content: str
|
|
) -> None:
|
|
file_path = os.path.join(ctx.cow_test_dir, filename)
|
|
assert os.path.exists(file_path), f"File {filename} not found"
|
|
with open(file_path) as f:
|
|
actual = f.read()
|
|
assert actual == content, f"Expected '{content}', got '{actual}'"
|
|
|
|
|
|
@then('the cow file "{filename}" in the original should have content "{content}"')
|
|
def step_cow_file_content(ctx: Context, filename: str, content: str) -> None:
|
|
file_path = os.path.join(ctx.cow_test_dir, filename)
|
|
with open(file_path) as f:
|
|
actual = f.read()
|
|
assert actual == content, f"Expected '{content}', got '{actual}'"
|
|
|
|
|
|
@then('the cow file "{filename}" should not exist in the original directory')
|
|
def step_cow_file_not_in_original(ctx: Context, filename: str) -> None:
|
|
file_path = os.path.join(ctx.cow_test_dir, filename)
|
|
assert not os.path.exists(file_path), f"File {filename} should not exist"
|
|
|
|
|
|
@then('the cow file "{filename}" should not exist in the sandbox')
|
|
def step_cow_file_not_in_sandbox(ctx: Context, filename: str) -> None:
|
|
if ctx.cow_sandbox.context:
|
|
file_path = os.path.join(ctx.cow_sandbox.context.sandbox_path, filename)
|
|
assert not os.path.exists(file_path)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then - protocol properties
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the cow sandbox_id should be a valid ULID")
|
|
def step_cow_valid_ulid(ctx: Context) -> None:
|
|
sid = ctx.cow_sandbox.sandbox_id
|
|
assert len(sid) == 26, f"Expected ULID length 26, got {len(sid)}"
|
|
|
|
|
|
@then('the cow sandbox status should be "{status}"')
|
|
def step_cow_status_is(ctx: Context, status: str) -> None:
|
|
assert ctx.cow_sandbox.status == SandboxStatus(status)
|
|
|
|
|
|
@then("the cow sandbox context should be None")
|
|
def step_cow_context_none(ctx: Context) -> None:
|
|
assert ctx.cow_sandbox.context is None
|