sand-boxer/tests/test_api.py

"""HTTP API v0 stub tests."""

from unittest.mock import patch

from fastapi.testclient import TestClient

from sandboxer.api.app import app
from sandboxer.models import (
    ActorType,
    Consumer,
    SandboxExecResult,
    SandboxState,
    SandboxStatus,
    SnapshotRecord,
)


def test_list_sandboxes_empty() -> None:
    with patch("sandboxer.api.app._manager") as mgr:
        mgr.list.return_value = []
        client = TestClient(app)
        assert client.get("/v1/sandboxes").json() == []


def test_get_sandbox_not_found() -> None:
    with patch("sandboxer.api.app._manager") as mgr:
        mgr.get.return_value = None
        client = TestClient(app)
        assert client.get("/v1/sandboxes/missing").status_code == 404


def test_create_sandbox() -> None:
    from datetime import UTC, datetime

    status = SandboxStatus(
        sandbox_id="abc12345",
        profile_id="profile.compose-e2e",
        extension_id="ext.compose-ssh",
        state=SandboxState.READY,
        consumer=Consumer(actor=ActorType.ADM, project="sand-boxer"),
        created_at=datetime.now(UTC),
        updated_at=datetime.now(UTC),
    )
    with patch("sandboxer.api.app._manager") as mgr:
        mgr.create.return_value = status
        client = TestClient(app)
        resp = client.post(
            "/v1/sandboxes",
            json={
                "profile": "profile.compose-e2e",
                "inputs": {"repo": "/tmp/x"},
                "consumer": {"actor": "adm", "project": "sand-boxer"},
            },
        )
        assert resp.status_code == 200
        assert resp.json()["sandbox_id"] == "abc12345"


def test_snapshot_sandbox() -> None:
    from datetime import UTC, datetime

    record = SnapshotRecord(
        snapshot_id="snap12345678",
        sandbox_id="abc12345",
        profile_id="profile.compose-checkpoint",
        extension_id="ext.compose-ssh",
        host="coulombcore",
        created_at=datetime.now(UTC),
    )
    with patch("sandboxer.api.app._manager") as mgr:
        mgr.snapshot.return_value = record
        client = TestClient(app)
        resp = client.post("/v1/sandboxes/abc12345/snapshot")
        assert resp.status_code == 200
        assert resp.json()["snapshot_id"] == "snap12345678"


def test_restore_snapshot() -> None:
    from datetime import UTC, datetime

    status = SandboxStatus(
        sandbox_id="restored1",
        profile_id="profile.compose-checkpoint",
        extension_id="ext.compose-ssh",
        state=SandboxState.READY,
        consumer=Consumer(actor=ActorType.ADM, project="sand-boxer"),
        created_at=datetime.now(UTC),
        updated_at=datetime.now(UTC),
    )
    with patch("sandboxer.api.app._manager") as mgr:
        mgr.restore.return_value = status
        client = TestClient(app)
        resp = client.post(
            "/v1/snapshots/snap12345678/restore",
            json={"consumer": {"actor": "adm", "project": "sand-boxer"}},
        )
        assert resp.status_code == 200
        assert resp.json()["sandbox_id"] == "restored1"


def test_recreate_sandbox() -> None:
    from datetime import UTC, datetime

    status = SandboxStatus(
        sandbox_id="new12345",
        profile_id="profile.compose-e2e",
        extension_id="ext.compose-ssh",
        state=SandboxState.READY,
        consumer=Consumer(actor=ActorType.ADM, project="sand-boxer"),
        created_at=datetime.now(UTC),
        updated_at=datetime.now(UTC),
    )
    with patch("sandboxer.api.app._manager") as mgr:
        mgr.recreate.return_value = status
        client = TestClient(app)
        resp = client.post("/v1/sandboxes/abc12345/recreate")
        assert resp.status_code == 200
        assert resp.json()["sandbox_id"] == "new12345"


def test_extend_ttl() -> None:
    from datetime import UTC, datetime

    now = datetime.now(UTC)
    status = SandboxStatus(
        sandbox_id="abc12345",
        profile_id="profile.compose-e2e",
        extension_id="ext.compose-ssh",
        state=SandboxState.READY,
        consumer=Consumer(actor=ActorType.ADM, project="sand-boxer"),
        ttl="2h",
        expires_at=now,
        created_at=now,
        updated_at=now,
        ready_at=now,
    )
    with patch("sandboxer.api.app._manager") as mgr:
        mgr.extend_ttl.return_value = status
        client = TestClient(app)
        resp = client.patch(
            "/v1/sandboxes/abc12345/ttl",
            json={"duration": "2h"},
        )
        assert resp.status_code == 200
        assert resp.json()["ttl"] == "2h"


def test_expire_sandboxes() -> None:
    from sandboxer.models import ExpireActionResult

    with patch("sandboxer.api.app._manager") as mgr:
        mgr.expire.return_value = [
            ExpireActionResult(sandbox_id="x", reason="ttl", action="dry-run")
        ]
        client = TestClient(app)
        resp = client.post("/v1/sandboxes/expire")
        assert resp.status_code == 200
        assert resp.json()[0]["action"] == "dry-run"


def test_exec_api_fails_closed_when_owner_token_unconfigured(monkeypatch) -> None:
    monkeypatch.delenv("SANDBOXER_EXEC_TOKEN", raising=False)
    client = TestClient(app)
    resp = client.post(
        "/v1/sandboxes/abc12345/exec",
        json={
            "command": ["true"],
            "consumer": {"actor": "agt", "project": "glas-harness", "run_id": "run-1"},
        },
    )
    assert resp.status_code == 503


def test_exec_api_requires_bearer_and_returns_execution(monkeypatch) -> None:
    from datetime import UTC, datetime

    monkeypatch.setenv("SANDBOXER_EXEC_TOKEN", "owner-capability")
    now = datetime.now(UTC)
    consumer = Consumer(actor="agt", project="glas-harness", run_id="run-1")
    result = SandboxExecResult(
        sandbox_id="abc12345",
        profile_id="profile.bwrap-local",
        extension_id="ext.bwrap",
        consumer=consumer,
        command_name="true",
        exit_code=0,
        duration_seconds=0.1,
        workspace_dir="/tmp/sandboxer-bwrap/abc12345",
        network_default="deny",
        started_at=now,
        completed_at=now,
    )
    payload = {
        "command": ["true"],
        "consumer": consumer.model_dump(mode="json"),
    }
    with patch("sandboxer.api.app._manager") as mgr:
        mgr.execute.return_value = result
        client = TestClient(app)
        denied = client.post(
            "/v1/sandboxes/abc12345/exec",
            json=payload,
            headers={"Authorization": "Bearer wrong"},
        )
        allowed = client.post(
            "/v1/sandboxes/abc12345/exec",
            json=payload,
            headers={"Authorization": "Bearer owner-capability"},
        )

    assert denied.status_code == 401
    assert allowed.status_code == 200
    assert allowed.json()["exit_code"] == 0