diff --git a/features/sandbox_manager_strategy_cast.feature b/features/sandbox_manager_strategy_cast.feature new file mode 100644 index 000000000..ad586cbfa --- /dev/null +++ b/features/sandbox_manager_strategy_cast.feature @@ -0,0 +1,18 @@ +Feature: Sandbox Manager strategy cast path + Exercises the cast("SandboxStrategyStr", boundary_resource.sandbox_strategy) + line in get_or_create_sandbox_for_resource, which is reached whenever the + boundary resource has a non-None sandbox_strategy. + + Background: + Given smcast a sandbox factory instance + And smcast a sandbox manager with the factory + + Scenario: get_or_create_sandbox_for_resource succeeds with SandboxStrategy.NONE + Given smcast a boundary resource with strategy "NONE" and location "/tmp/smcast-none" + When smcast I call get_or_create_sandbox_for_resource for plan "plan-smcast-1" + Then smcast a sandbox is returned without error + + Scenario: get_or_create_sandbox_for_resource succeeds with SandboxStrategy.NONE on a second plan + Given smcast a boundary resource with strategy "NONE" and location "/tmp/smcast-none-2" + When smcast I call get_or_create_sandbox_for_resource for plan "plan-smcast-2" + Then smcast a sandbox is returned without error diff --git a/features/steps/sandbox_manager_strategy_cast_steps.py b/features/steps/sandbox_manager_strategy_cast_steps.py new file mode 100644 index 000000000..14aa2d839 --- /dev/null +++ b/features/steps/sandbox_manager_strategy_cast_steps.py @@ -0,0 +1,165 @@ +"""Step definitions for sandbox manager strategy cast coverage. + +Exercises the cast("SandboxStrategyStr", boundary_resource.sandbox_strategy) +line in get_or_create_sandbox_for_resource, which is reached whenever the +boundary resource has a non-None sandbox_strategy. +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import 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 + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +# Valid ULID strings for test resources +_ULID_CHILD = "01ARZ3NDEKTSV4RRFFQ69G5FAV" +_ULID_BOUNDARY = "01BRZ3NDEKTSV4RRFFQ69G5FAV" + +_STRATEGY_MAP: dict[str, SandboxStrategy] = { + "NONE": SandboxStrategy.NONE, + "GIT_WORKTREE": SandboxStrategy.GIT_WORKTREE, + "COPY_ON_WRITE": SandboxStrategy.COPY_ON_WRITE, + "TRANSACTION_ROLLBACK": SandboxStrategy.TRANSACTION_ROLLBACK, + "SNAPSHOT": SandboxStrategy.SNAPSHOT, + "OVERLAY": SandboxStrategy.OVERLAY, +} + + +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("smcast a sandbox factory instance") +def step_smcast_given_factory(context: Any) -> None: + context.smcast_factory = SandboxFactory() + + +@given("smcast a sandbox manager with the factory") +def step_smcast_given_manager(context: Any) -> None: + context.smcast_manager = SandboxManager( + factory=context.smcast_factory, cleanup_on_exit=False + ) + context.smcast_error = None + context.smcast_sandbox = None + context.smcast_boundary_resource = None + context.smcast_child_resource = None + context.smcast_registry = None + + +# --------------------------------------------------------------------------- +# Given steps +# --------------------------------------------------------------------------- + + +@given( + 'smcast a boundary resource with strategy "{strategy}" and location "{location}"' +) +def step_smcast_given_boundary_resource( + context: Any, strategy: str, location: str +) -> None: + """Set up a boundary resource with the given strategy and location.""" + sandbox_strategy = _STRATEGY_MAP[strategy] + + boundary_resource = _make_resource( + resource_id=_ULID_BOUNDARY, + location=location, + sandbox_strategy=sandbox_strategy, + sandboxable=True, + ) + + child_resource = _make_resource( + resource_id=_ULID_CHILD, + location="/tmp/smcast-child", + sandbox_strategy=None, + sandboxable=False, + parents=[_ULID_BOUNDARY], + ) + + context.smcast_boundary_resource = boundary_resource + context.smcast_child_resource = child_resource + context.smcast_registry = { + _ULID_BOUNDARY: boundary_resource, + _ULID_CHILD: child_resource, + } + + +# --------------------------------------------------------------------------- +# When steps +# --------------------------------------------------------------------------- + + +@when('smcast I call get_or_create_sandbox_for_resource for plan "{plan_id}"') +def step_smcast_when_call_for_resource(context: Any, plan_id: str) -> None: + """Call get_or_create_sandbox_for_resource with the prepared boundary resource. + + The boundary cache is patched to return the boundary resource directly, + bypassing DAG traversal. + """ + context.smcast_error = None + context.smcast_sandbox = None + + patcher = patch.object( + context.smcast_manager._boundary_cache, + "get_boundary", + return_value=context.smcast_boundary_resource, + ) + patcher.start() + context.add_cleanup(patcher.stop) + + try: + context.smcast_sandbox = ( + context.smcast_manager.get_or_create_sandbox_for_resource( + plan_id=plan_id, + resource=context.smcast_child_resource, + resource_registry=context.smcast_registry, + ) + ) + except Exception as exc: + context.smcast_error = exc + + +# --------------------------------------------------------------------------- +# Then steps +# --------------------------------------------------------------------------- + + +@then("smcast a sandbox is returned without error") +def step_smcast_then_sandbox_returned(context: Any) -> None: + assert context.smcast_error is None, f"Unexpected error: {context.smcast_error}" + assert context.smcast_sandbox is not None, ( + "Expected a sandbox to be returned, but got None" + ) diff --git a/src/cleveragents/infrastructure/sandbox/manager.py b/src/cleveragents/infrastructure/sandbox/manager.py index c9b9816b1..d72d46289 100644 --- a/src/cleveragents/infrastructure/sandbox/manager.py +++ b/src/cleveragents/infrastructure/sandbox/manager.py @@ -18,6 +18,7 @@ import logging import threading from collections.abc import Mapping from datetime import datetime +from typing import cast from cleveragents.domain.models.core.resource import Resource from cleveragents.infrastructure.sandbox.boundary import ( @@ -615,7 +616,7 @@ class SandboxManager: # Determine strategy from the boundary resource. strategy: SandboxStrategyStr = "none" if boundary_resource.sandbox_strategy is not None: - strategy = boundary_resource.sandbox_strategy # type: ignore[assignment] + strategy = cast("SandboxStrategyStr", boundary_resource.sandbox_strategy) # Determine location from the boundary resource. original_path = boundary_resource.location or ""