forked from HAL9000/cleveragents-core
freemo
31472b5413
Add Behave feature/step pairs that exercise uncovered branches across handlers, LSP, CLI, and service layers to reach the coverage gate. ISSUES CLOSED: #1232
397 lines
13 KiB
Python
397 lines
13 KiB
Python
"""Step definitions for sandbox manager coverage round 3.
|
|
|
|
Covers uncovered lines in SandboxManager:
|
|
- TransactionSandbox branch in commit_all (lines 287-295)
|
|
- resolve_sandbox_key validation (lines 567, 569)
|
|
- get_or_create_sandbox_for_resource validation (lines 606, 608, 610, 623)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from typing import Any
|
|
from unittest.mock import MagicMock, PropertyMock, patch
|
|
|
|
from behave import given, then, when
|
|
|
|
from cleveragents.domain.models.core.resource import (
|
|
PhysVirt,
|
|
Resource,
|
|
ResourceCapabilities,
|
|
SandboxStrategy,
|
|
)
|
|
from cleveragents.infrastructure.sandbox.factory import SandboxFactory
|
|
from cleveragents.infrastructure.sandbox.manager import SandboxManager
|
|
from cleveragents.infrastructure.sandbox.protocol import (
|
|
CommitResult,
|
|
SandboxContext,
|
|
SandboxError,
|
|
SandboxStatus,
|
|
)
|
|
from cleveragents.infrastructure.sandbox.transaction_sandbox import TransactionSandbox
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# Valid ULID strings for test resources
|
|
_ULID_CHILD = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
|
|
_ULID_BOUNDARY = "01BRZ3NDEKTSV4RRFFQ69G5FAV"
|
|
|
|
|
|
def _make_mock_sandbox(
|
|
sandbox_id: str = "sb-smcov3-001",
|
|
status: SandboxStatus = SandboxStatus.ACTIVE,
|
|
resource_id: str = "res-smcov3",
|
|
original_path: str = "/tmp/smcov3",
|
|
plan_id: str = "plan-smcov3",
|
|
) -> MagicMock:
|
|
"""Create a mock sandbox satisfying the Sandbox protocol."""
|
|
sb = MagicMock()
|
|
sb.sandbox_id = sandbox_id
|
|
type(sb).status = PropertyMock(return_value=status)
|
|
sb.context = SandboxContext(
|
|
sandbox_id=sandbox_id,
|
|
sandbox_path=original_path,
|
|
original_path=original_path,
|
|
resource_id=resource_id,
|
|
plan_id=plan_id,
|
|
created_at=datetime.now(),
|
|
)
|
|
sb.create.return_value = sb.context
|
|
sb.commit.return_value = CommitResult(
|
|
sandbox_id=sandbox_id,
|
|
success=True,
|
|
timestamp=datetime.now(),
|
|
)
|
|
sb.rollback.return_value = None
|
|
sb.cleanup.return_value = None
|
|
return sb
|
|
|
|
|
|
def _make_transaction_sandbox_mock(
|
|
sandbox_id: str = "sb-txn-001",
|
|
status: SandboxStatus = SandboxStatus.ACTIVE,
|
|
resource_id: str = "res-txn",
|
|
plan_id: str = "plan-txn",
|
|
) -> MagicMock:
|
|
"""Create a mock that passes isinstance(sb, TransactionSandbox)."""
|
|
sb = MagicMock(spec=TransactionSandbox)
|
|
sb.sandbox_id = sandbox_id
|
|
type(sb).status = PropertyMock(return_value=status)
|
|
sb.context = SandboxContext(
|
|
sandbox_id=sandbox_id,
|
|
sandbox_path="/tmp/txn",
|
|
original_path="/tmp/txn",
|
|
resource_id=resource_id,
|
|
plan_id=plan_id,
|
|
created_at=datetime.now(),
|
|
)
|
|
sb.create.return_value = sb.context
|
|
sb.commit.return_value = CommitResult(
|
|
sandbox_id=sandbox_id,
|
|
success=True,
|
|
timestamp=datetime.now(),
|
|
)
|
|
sb.rollback.return_value = None
|
|
sb.cleanup.return_value = None
|
|
return sb
|
|
|
|
|
|
def _make_resource(
|
|
resource_id: str = _ULID_CHILD,
|
|
location: str | None = "/tmp/repo",
|
|
sandbox_strategy: SandboxStrategy | None = SandboxStrategy.GIT_WORKTREE,
|
|
sandboxable: bool = True,
|
|
parents: list[str] | None = None,
|
|
) -> Resource:
|
|
"""Create a Resource domain object for testing."""
|
|
return Resource(
|
|
resource_id=resource_id,
|
|
resource_type_name="git-checkout",
|
|
classification=PhysVirt.PHYSICAL,
|
|
location=location,
|
|
sandbox_strategy=sandbox_strategy,
|
|
capabilities=ResourceCapabilities(sandboxable=sandboxable),
|
|
parents=parents or [],
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("smcov3 a sandbox factory instance")
|
|
def step_smcov3_given_factory(context: Any) -> None:
|
|
context.smcov3_factory = SandboxFactory()
|
|
|
|
|
|
@given("smcov3 a sandbox manager with the factory")
|
|
def step_smcov3_given_manager(context: Any) -> None:
|
|
context.smcov3_manager = SandboxManager(
|
|
factory=context.smcov3_factory, cleanup_on_exit=False
|
|
)
|
|
context.smcov3_error = None
|
|
context.smcov3_commit_results = None
|
|
context.smcov3_sandbox_refs = {}
|
|
context.smcov3_txn_mock_refs = {}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Given steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given(
|
|
'smcov3 a sandbox exists for plan "{plan_id}" resource "{res_id}" '
|
|
'path "{path}" strategy "{strategy}"'
|
|
)
|
|
def step_smcov3_given_sandbox_exists(
|
|
context: Any, plan_id: str, res_id: str, path: str, strategy: str
|
|
) -> None:
|
|
sandbox = context.smcov3_manager.get_or_create_sandbox(
|
|
plan_id=plan_id,
|
|
resource_id=res_id,
|
|
original_path=path,
|
|
sandbox_strategy=strategy,
|
|
)
|
|
context.smcov3_sandbox_refs[f"{plan_id}:{res_id}"] = sandbox
|
|
|
|
|
|
@given(
|
|
'smcov3 the sandbox for plan "{plan_id}" resource "{res_id}" '
|
|
"is replaced with a transaction sandbox mock"
|
|
)
|
|
def step_smcov3_given_replace_with_txn_mock(
|
|
context: Any, plan_id: str, res_id: str
|
|
) -> None:
|
|
"""Replace sandbox with a mock that passes isinstance(sb, TransactionSandbox)."""
|
|
existing = context.smcov3_manager.get_sandbox(plan_id, res_id)
|
|
assert existing is not None, (
|
|
f"No sandbox found for plan={plan_id} resource={res_id}"
|
|
)
|
|
mock_sb = _make_transaction_sandbox_mock(
|
|
sandbox_id=existing.sandbox_id,
|
|
status=SandboxStatus.ACTIVE,
|
|
resource_id=res_id,
|
|
plan_id=plan_id,
|
|
)
|
|
context.smcov3_manager._active_sandboxes[plan_id][res_id] = mock_sb
|
|
context.smcov3_txn_mock_refs[f"{plan_id}:{res_id}"] = mock_sb
|
|
|
|
|
|
@given(
|
|
'smcov3 the sandbox for plan "{plan_id}" resource "{res_id}" '
|
|
"is replaced with a committable mock"
|
|
)
|
|
def step_smcov3_given_replace_with_committable(
|
|
context: Any, plan_id: str, res_id: str
|
|
) -> None:
|
|
"""Replace sandbox with a standard mock that commits and can be rolled back."""
|
|
existing = context.smcov3_manager.get_sandbox(plan_id, res_id)
|
|
assert existing is not None, (
|
|
f"No sandbox found for plan={plan_id} resource={res_id}"
|
|
)
|
|
mock_sb = _make_mock_sandbox(
|
|
sandbox_id=existing.sandbox_id,
|
|
status=SandboxStatus.ACTIVE,
|
|
resource_id=res_id,
|
|
plan_id=plan_id,
|
|
)
|
|
context.smcov3_manager._active_sandboxes[plan_id][res_id] = mock_sb
|
|
context.smcov3_sandbox_refs[f"{plan_id}:{res_id}"] = mock_sb
|
|
|
|
|
|
@given(
|
|
'smcov3 the sandbox for plan "{plan_id}" resource "{res_id}" will fail on commit'
|
|
)
|
|
def step_smcov3_given_commit_fails(context: Any, plan_id: str, res_id: str) -> None:
|
|
"""Make the sandbox's commit method raise SandboxError."""
|
|
existing = context.smcov3_manager.get_sandbox(plan_id, res_id)
|
|
assert existing is not None
|
|
mock_sb = _make_mock_sandbox(
|
|
sandbox_id=existing.sandbox_id,
|
|
status=SandboxStatus.ACTIVE,
|
|
resource_id=res_id,
|
|
plan_id=plan_id,
|
|
)
|
|
mock_sb.commit.side_effect = SandboxError("smcov3 commit failed")
|
|
context.smcov3_manager._active_sandboxes[plan_id][res_id] = mock_sb
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('smcov3 I commit all sandboxes for plan "{plan_id}"')
|
|
def step_smcov3_when_commit_all(context: Any, plan_id: str) -> None:
|
|
context.smcov3_error = None
|
|
try:
|
|
context.smcov3_commit_results = context.smcov3_manager.commit_all(plan_id)
|
|
except Exception as exc:
|
|
context.smcov3_error = exc
|
|
|
|
|
|
@when("smcov3 I call resolve_sandbox_key with None resource")
|
|
def step_smcov3_when_resolve_key_none_resource(context: Any) -> None:
|
|
context.smcov3_error = None
|
|
try:
|
|
context.smcov3_manager.resolve_sandbox_key(
|
|
resource=None,
|
|
resource_registry={},
|
|
)
|
|
except Exception as exc:
|
|
context.smcov3_error = exc
|
|
|
|
|
|
@when("smcov3 I call resolve_sandbox_key with None resource_registry")
|
|
def step_smcov3_when_resolve_key_none_registry(context: Any) -> None:
|
|
context.smcov3_error = None
|
|
resource = _make_resource()
|
|
try:
|
|
context.smcov3_manager.resolve_sandbox_key(
|
|
resource=resource,
|
|
resource_registry=None,
|
|
)
|
|
except Exception as exc:
|
|
context.smcov3_error = exc
|
|
|
|
|
|
@when("smcov3 I call get_or_create_sandbox_for_resource with empty plan_id")
|
|
def step_smcov3_when_for_resource_empty_plan(context: Any) -> None:
|
|
context.smcov3_error = None
|
|
resource = _make_resource()
|
|
try:
|
|
context.smcov3_manager.get_or_create_sandbox_for_resource(
|
|
plan_id="",
|
|
resource=resource,
|
|
resource_registry={resource.resource_id: resource},
|
|
)
|
|
except Exception as exc:
|
|
context.smcov3_error = exc
|
|
|
|
|
|
@when("smcov3 I call get_or_create_sandbox_for_resource with None resource")
|
|
def step_smcov3_when_for_resource_none_resource(context: Any) -> None:
|
|
context.smcov3_error = None
|
|
try:
|
|
context.smcov3_manager.get_or_create_sandbox_for_resource(
|
|
plan_id="plan-valid",
|
|
resource=None,
|
|
resource_registry={},
|
|
)
|
|
except Exception as exc:
|
|
context.smcov3_error = exc
|
|
|
|
|
|
@when("smcov3 I call get_or_create_sandbox_for_resource with None resource_registry")
|
|
def step_smcov3_when_for_resource_none_registry(context: Any) -> None:
|
|
context.smcov3_error = None
|
|
resource = _make_resource()
|
|
try:
|
|
context.smcov3_manager.get_or_create_sandbox_for_resource(
|
|
plan_id="plan-valid",
|
|
resource=resource,
|
|
resource_registry=None,
|
|
)
|
|
except Exception as exc:
|
|
context.smcov3_error = exc
|
|
|
|
|
|
@when(
|
|
"smcov3 I call get_or_create_sandbox_for_resource with a boundary that has no location"
|
|
)
|
|
def step_smcov3_when_for_resource_no_location(context: Any) -> None:
|
|
"""Call get_or_create_sandbox_for_resource where the boundary resource has no location.
|
|
|
|
We create a boundary resource (sandboxable, with a strategy) but location=None,
|
|
and mock the boundary cache to return it directly.
|
|
"""
|
|
context.smcov3_error = None
|
|
|
|
boundary_resource = _make_resource(
|
|
resource_id=_ULID_BOUNDARY,
|
|
location=None,
|
|
sandbox_strategy=SandboxStrategy.GIT_WORKTREE,
|
|
sandboxable=True,
|
|
)
|
|
|
|
child_resource = _make_resource(
|
|
resource_id=_ULID_CHILD,
|
|
location="/tmp/child",
|
|
sandbox_strategy=None,
|
|
sandboxable=False,
|
|
parents=[_ULID_BOUNDARY],
|
|
)
|
|
|
|
registry = {
|
|
_ULID_BOUNDARY: boundary_resource,
|
|
_ULID_CHILD: child_resource,
|
|
}
|
|
|
|
# Patch the boundary cache to return the boundary resource directly
|
|
patcher = patch.object(
|
|
context.smcov3_manager._boundary_cache,
|
|
"get_boundary",
|
|
return_value=boundary_resource,
|
|
)
|
|
patcher.start()
|
|
context.add_cleanup(patcher.stop)
|
|
|
|
try:
|
|
context.smcov3_manager.get_or_create_sandbox_for_resource(
|
|
plan_id="plan-loc",
|
|
resource=child_resource,
|
|
resource_registry=registry,
|
|
)
|
|
except Exception as exc:
|
|
context.smcov3_error = exc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("smcov3 the commit results should contain {count:d} successful result")
|
|
def step_smcov3_then_commit_results_count(context: Any, count: int) -> None:
|
|
assert context.smcov3_commit_results is not None, "No commit results available"
|
|
successful = [r for r in context.smcov3_commit_results if r.success]
|
|
assert len(successful) == count, (
|
|
f"Expected {count} successful result(s), got {len(successful)}"
|
|
)
|
|
|
|
|
|
@then("smcov3 no error should have been raised")
|
|
def step_smcov3_then_no_error(context: Any) -> None:
|
|
assert context.smcov3_error is None, f"Unexpected error: {context.smcov3_error}"
|
|
|
|
|
|
@then('smcov3 a ValueError should have been raised with message "{message}"')
|
|
def step_smcov3_then_value_error(context: Any, message: str) -> None:
|
|
assert context.smcov3_error is not None, (
|
|
"Expected a ValueError but no error was raised"
|
|
)
|
|
assert isinstance(context.smcov3_error, ValueError), (
|
|
f"Expected ValueError, got {type(context.smcov3_error).__name__}: "
|
|
f"{context.smcov3_error}"
|
|
)
|
|
assert message in str(context.smcov3_error), (
|
|
f"Expected '{message}' in '{context.smcov3_error}'"
|
|
)
|
|
|
|
|
|
@then(
|
|
'smcov3 the transaction sandbox mock for plan "{plan_id}" resource "{res_id}" '
|
|
"should not have been committed"
|
|
)
|
|
def step_smcov3_then_txn_mock_not_committed(
|
|
context: Any, plan_id: str, res_id: str
|
|
) -> None:
|
|
key = f"{plan_id}:{res_id}"
|
|
mock_sb = context.smcov3_txn_mock_refs.get(key)
|
|
assert mock_sb is not None, f"No transaction sandbox mock found for {key}"
|
|
mock_sb.commit.assert_not_called()
|