master/test/test_docker_runtime.py

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

CHAT_ID = '123e4567-e89b-12d3-a456-426614174000'
NON_CANONICAL_CHAT_ID = '123E4567E89B12D3A456426614174000'


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_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 = 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=NON_CANONICAL_CHAT_ID,
        created_at=created_at,
        expires_at=expires_at,
    )

    assert session.session_id == 'session-123'
    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' / 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': 'session-123',
        'chat_id': 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' / 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 = DockerSandboxRuntime(config, FakeDockerClient(containers))

    with pytest.raises(SandboxStartError) as excinfo:
        runtime.create(
            session_id='session-123',
            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 == 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 = 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'


@pytest.mark.parametrize('chat_id', ['.', 'a/..', 'x/../y'])
def test_runtime_create_rejects_non_uuid_chat_id(tmp_path: Path, chat_id: str) -> 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=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 == chat_id
    assert containers.run_calls == []