from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any, TypedDict from uuid import UUID import pytest from docker import DockerClient from docker.errors import DockerException, NotFound from docker.types import Mount from adapter.config.model import SandboxConfig from adapter.docker.runtime import DockerSandboxRuntime from adapter.observability.noop import NoopMetrics, NoopTracer from domain.error import SandboxError, SandboxStartError from domain.sandbox import SandboxSession, SandboxStatus CHAT_ID = UUID('123e4567-e89b-12d3-a456-426614174000') NON_CANONICAL_CHAT_ID = '123E4567E89B12D3A456426614174000' SESSION_ID = UUID('00000000-0000-0000-0000-000000000010') class FakeContainer: def __init__(self, container_id: str) -> None: self.id = container_id self.stop_calls = 0 def stop(self) -> None: self.stop_calls += 1 class FakeListedContainer(FakeContainer): def __init__( self, container_id: str, *, labels: dict[str, str], created_at: str, ) -> None: super().__init__(container_id) self.labels = labels self.attrs = {'Created': created_at} class RunKwargs(TypedDict): detach: bool labels: dict[str, str] mounts: list[Mount] class RunCall(TypedDict): args: tuple[str] kwargs: RunKwargs class FakeContainers: def __init__(self, run_result: FakeContainer | None = None) -> None: self.run_calls: list[RunCall] = [] self.get_calls: list[str] = [] self.list_calls: list[dict[str, object]] = [] self.run_result = run_result or FakeContainer('container-123') self.get_result: FakeContainer | Exception | None = None self.list_result: list[object] = [] def run( self, image: str, *, detach: bool, labels: dict[str, str], mounts: list[Mount], ) -> FakeContainer: self.run_calls.append( { 'args': (image,), 'kwargs': { 'detach': detach, 'labels': labels, 'mounts': mounts, }, } ) return self.run_result def get(self, container_id: str) -> FakeContainer: self.get_calls.append(container_id) if isinstance(self.get_result, Exception): raise self.get_result if self.get_result is None: raise AssertionError('missing get result') return self.get_result def list(self, *, filters: dict[str, list[str]]) -> list[object]: self.list_calls.append({'filters': filters}) return self.list_result class FakeDockerClient(DockerClient): def __init__(self, containers: FakeContainers) -> None: self._containers = containers @property def containers(self) -> Any: return self._containers def build_config(tmp_path: Path) -> SandboxConfig: return SandboxConfig( image='sandbox:latest', ttl_seconds=300, cleanup_interval_seconds=60, chats_root=str(tmp_path / 'chats'), dependencies_host_path=str(tmp_path / 'dependencies'), lambda_tools_host_path=str(tmp_path / 'lambda-tools'), chat_mount_path='/workspace/chat', dependencies_mount_path='/workspace/dependencies', lambda_tools_mount_path='/workspace/lambda-tools', ) def build_runtime( config: SandboxConfig, containers: FakeContainers, ) -> DockerSandboxRuntime: return DockerSandboxRuntime( config, FakeDockerClient(containers), NoopMetrics(), NoopTracer(), ) def test_runtime_create_applies_mount_policy_and_labels_with_canonical_chat_id( tmp_path: Path, ) -> None: config = build_config(tmp_path) (tmp_path / 'dependencies').mkdir() (tmp_path / 'lambda-tools').mkdir() containers = FakeContainers() runtime = build_runtime(config, containers) created_at = datetime(2026, 4, 2, 12, 0, tzinfo=UTC) expires_at = created_at + timedelta(minutes=5) session = runtime.create( session_id=SESSION_ID, chat_id=UUID(NON_CANONICAL_CHAT_ID), created_at=created_at, expires_at=expires_at, ) assert session.session_id == SESSION_ID assert session.chat_id == CHAT_ID assert session.container_id == 'container-123' assert session.status is SandboxStatus.RUNNING assert session.created_at == created_at assert session.expires_at == expires_at assert (tmp_path / 'chats' / str(CHAT_ID)).is_dir() call = containers.run_calls[0] assert call['args'] == ('sandbox:latest',) assert call['kwargs']['detach'] is True assert call['kwargs']['labels'] == { 'session_id': str(SESSION_ID), 'chat_id': str(CHAT_ID), 'expires_at': expires_at.isoformat(), } mounts = call['kwargs']['mounts'] assert [dict(mount) for mount in mounts] == [ { 'Target': '/workspace/chat', 'Source': str((tmp_path / 'chats' / str(CHAT_ID)).resolve(strict=False)), 'Type': 'bind', 'ReadOnly': False, }, { 'Target': '/workspace/dependencies', 'Source': str((tmp_path / 'dependencies').resolve(strict=False)), 'Type': 'bind', 'ReadOnly': True, }, { 'Target': '/workspace/lambda-tools', 'Source': str((tmp_path / 'lambda-tools').resolve(strict=False)), 'Type': 'bind', 'ReadOnly': True, }, ] def test_runtime_create_raises_start_error_when_container_id_is_missing( tmp_path: Path, ) -> None: config = build_config(tmp_path) (tmp_path / 'dependencies').mkdir() (tmp_path / 'lambda-tools').mkdir() containers = FakeContainers(run_result=FakeContainer('')) runtime = build_runtime(config, containers) with pytest.raises(SandboxStartError) as excinfo: runtime.create( session_id=SESSION_ID, chat_id=CHAT_ID, created_at=datetime(2026, 4, 2, 12, 0, tzinfo=UTC), expires_at=datetime(2026, 4, 2, 12, 5, tzinfo=UTC), ) assert str(excinfo.value) == 'sandbox_start_failed' assert excinfo.value.chat_id == str(CHAT_ID) def test_runtime_stop_ignores_missing_container(tmp_path: Path) -> None: config = build_config(tmp_path) containers = FakeContainers() containers.get_result = NotFound('missing') runtime = build_runtime(config, containers) runtime.stop('container-123') assert containers.get_calls == ['container-123'] def test_runtime_stop_wraps_docker_errors(tmp_path: Path) -> None: config = build_config(tmp_path) containers = FakeContainers() containers.get_result = DockerException('boom') runtime = build_runtime(config, containers) with pytest.raises(SandboxError) as excinfo: runtime.stop('container-123') assert str(excinfo.value) == 'sandbox_stop_failed' def test_runtime_list_active_sessions_reads_valid_labeled_containers( tmp_path: Path, ) -> None: config = build_config(tmp_path) containers = FakeContainers() expires_at = datetime(2026, 4, 2, 12, 5, tzinfo=UTC) containers.list_result = [ FakeListedContainer( 'container-123', labels={ 'session_id': str(SESSION_ID), 'chat_id': str(CHAT_ID), 'expires_at': expires_at.isoformat(), }, created_at='2026-04-02T12:00:00Z', ), FakeListedContainer( 'container-bad', labels={ 'chat_id': str(CHAT_ID), 'expires_at': expires_at.isoformat(), }, created_at='2026-04-02T12:01:00Z', ), ] runtime = build_runtime(config, containers) sessions = runtime.list_active_sessions() assert sessions == [ SandboxSession( session_id=SESSION_ID, chat_id=CHAT_ID, container_id='container-123', status=SandboxStatus.RUNNING, created_at=datetime(2026, 4, 2, 12, 0, tzinfo=UTC), expires_at=expires_at, ) ] assert containers.list_calls == [ {'filters': {'label': ['session_id', 'chat_id', 'expires_at']}} ]