Add meta-framework spec, pydantic schemas, profile/extension YAML, extension registry, ext.compose-ssh backend, SandboxManager with State Hub events, CLI commands, integration docs, capability registry entry, and compose-e2e runbook. Nine unit tests pass. T10 remote smoke test remains for operator.
84 lines
No EOL
2.6 KiB
Python
84 lines
No EOL
2.6 KiB
Python
"""SandboxManager unit tests with mocked backend."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from sandboxer.core.manager import SandboxManager
|
|
from sandboxer.lifecycle.store import SandboxStore
|
|
from sandboxer.models import ActorType, Consumer, SandboxCreateRequest, SandboxState, SandboxStatus
|
|
|
|
|
|
class FakeBackend:
|
|
def provision(self, profile, inputs, host):
|
|
return {
|
|
"sandbox_id": "test1234",
|
|
"host": host,
|
|
"remote_dir": "/tmp/sandboxer/test1234",
|
|
"compose_project": "sbx-e2e-test1234",
|
|
"compose_file": "docker-compose.yml",
|
|
"ssh_user": "root",
|
|
}
|
|
|
|
def wait_ready(self, handle):
|
|
return {
|
|
"ssh": f"root@{handle['host']}",
|
|
"remote_dir": handle["remote_dir"],
|
|
"compose_project": handle["compose_project"],
|
|
"host": handle["host"],
|
|
}
|
|
|
|
def teardown(self, handle):
|
|
return {
|
|
"compose_removed": "True",
|
|
"remote_dir_removed": "True",
|
|
"remote_dir": handle["remote_dir"],
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def store(tmp_path: Path) -> SandboxStore:
|
|
return SandboxStore(path=tmp_path / "sandboxes.json")
|
|
|
|
|
|
def test_create_and_destroy(store: SandboxStore) -> None:
|
|
manager = SandboxManager(store=store)
|
|
request = SandboxCreateRequest(
|
|
profile="profile.compose-e2e",
|
|
inputs={"repo": "/tmp/repo"},
|
|
consumer=Consumer(actor=ActorType.ADM, project="sand-boxer"),
|
|
)
|
|
|
|
with (
|
|
patch("sandboxer.core.manager.resolve_backend", return_value=FakeBackend()),
|
|
patch("sandboxer.core.manager.emit_lifecycle_event", return_value=None),
|
|
patch("sandboxer.core.manager.resolve_host", return_value="coulombcore"),
|
|
):
|
|
status = manager.create(request)
|
|
assert status.state == SandboxState.READY
|
|
assert status.sandbox_id == "test1234"
|
|
|
|
destroyed = manager.destroy(status.sandbox_id)
|
|
assert destroyed.state == SandboxState.DESTROYED
|
|
|
|
|
|
def test_destroy_idempotent(store: SandboxStore) -> None:
|
|
now = datetime.now(UTC)
|
|
status = SandboxStatus(
|
|
sandbox_id="gone1234",
|
|
profile_id="profile.compose-e2e",
|
|
extension_id="ext.compose-ssh",
|
|
state=SandboxState.DESTROYED,
|
|
consumer=Consumer(actor=ActorType.ADM, project="sand-boxer"),
|
|
created_at=now,
|
|
updated_at=now,
|
|
destroyed_at=now,
|
|
)
|
|
store.save(status)
|
|
manager = SandboxManager(store=store)
|
|
result = manager.destroy("gone1234")
|
|
assert result.state == SandboxState.DESTROYED |