ref #9: [feat] add tests

test/test_docker_runtime.py

@@ -0,0 +1,212 @@
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from typing import Any, TypedDict
+
+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 domain.error import SandboxError, SandboxStartError
+from domain.sandbox import SandboxStatus
+
+
+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 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.run_result = run_result or FakeContainer('container-123')
+        self.get_result: FakeContainer | Exception | None = None
+
+    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
+
+
+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 test_runtime_create_applies_mount_policy_and_labels(tmp_path: Path) -> None:
+    config = build_config(tmp_path)
+    (tmp_path / 'dependencies').mkdir()
+    (tmp_path / 'lambda-tools').mkdir()
+    containers = FakeContainers()
+    runtime = DockerSandboxRuntime(config, FakeDockerClient(containers))
+    created_at = datetime(2026, 4, 2, 12, 0, tzinfo=UTC)
+    expires_at = created_at + timedelta(minutes=5)
+
+    session = runtime.create(
+        session_id='session-123',
+        chat_id='chat-123',
+        created_at=created_at,
+        expires_at=expires_at,
+    )
+
+    assert session.session_id == 'session-123'
+    assert session.chat_id == 'chat-123'
+    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' / 'chat-123').is_dir()
+
+    call = containers.run_calls[0]
+    assert call['args'] == ('sandbox:latest',)
+    assert call['kwargs']['detach'] is True
+    assert call['kwargs']['labels'] == {
+        'session_id': 'session-123',
+        'chat_id': 'chat-123',
+        'expires_at': expires_at.isoformat(),
+    }
+
+    mounts = call['kwargs']['mounts']
+    assert [dict(mount) for mount in mounts] == [
+        {
+            'Target': '/workspace/chat',
+            'Source': str((tmp_path / 'chats' / 'chat-123').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 = DockerSandboxRuntime(config, FakeDockerClient(containers))
+
+    with pytest.raises(SandboxStartError) as excinfo:
+        runtime.create(
+            session_id='session-123',
+            chat_id='chat-123',
+            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 == 'chat-123'
+
+
+def test_runtime_stop_ignores_missing_container(tmp_path: Path) -> None:
+    config = build_config(tmp_path)
+    containers = FakeContainers()
+    containers.get_result = NotFound('missing')
+    runtime = DockerSandboxRuntime(config, FakeDockerClient(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 = DockerSandboxRuntime(config, FakeDockerClient(containers))
+
+    with pytest.raises(SandboxError) as excinfo:
+        runtime.stop('container-123')
+
+    assert str(excinfo.value) == 'sandbox_stop_failed'
+
+
+def test_runtime_create_rejects_chat_path_traversal(tmp_path: Path) -> None:
+    config = build_config(tmp_path)
+    (tmp_path / 'dependencies').mkdir()
+    (tmp_path / 'lambda-tools').mkdir()
+    containers = FakeContainers()
+    runtime = DockerSandboxRuntime(config, FakeDockerClient(containers))
+
+    with pytest.raises(SandboxStartError) as excinfo:
+        runtime.create(
+            session_id='session-123',
+            chat_id='../escape',
+            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 == '../escape'
+    assert containers.run_calls == []