generated from coulomb/repo-seed
tegwick
df658e7ef9
Add TTL parser, expires_at on create, extend_ttl and expire/reap APIs, activity-core integration doc, repo classification, registry refresh, HTTP parity, and 69 tests.
265 lines
8.6 KiB
Python
265 lines
8.6 KiB
Python
"""TTL parsing, extend, and expire tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from sandboxer.core.manager import SandboxManager
|
|
from sandboxer.lifecycle.expire import find_expire_candidates
|
|
from sandboxer.lifecycle.store import SandboxStore
|
|
from sandboxer.lifecycle.ttl import (
|
|
cap_duration,
|
|
extend_expires_at,
|
|
format_timedelta,
|
|
is_idle_expired,
|
|
is_past_expiry,
|
|
parse_duration,
|
|
resolve_initial_ttl,
|
|
)
|
|
from sandboxer.models import (
|
|
ActorType,
|
|
Consumer,
|
|
Profile,
|
|
Reachability,
|
|
SandboxCreateRequest,
|
|
SandboxState,
|
|
SandboxStatus,
|
|
)
|
|
|
|
|
|
def _profile(**ttl_overrides) -> Profile:
|
|
ttl_data = {"default": "4h", "max": "24h", "idle_reap": None}
|
|
ttl_data.update(ttl_overrides)
|
|
return Profile.model_validate(
|
|
{
|
|
"id": "profile.compose-e2e",
|
|
"version": "1.0.0",
|
|
"extension": "ext.compose-ssh",
|
|
"ttl": ttl_data,
|
|
}
|
|
)
|
|
|
|
|
|
def test_parse_duration_units() -> None:
|
|
assert parse_duration("30m") == timedelta(minutes=30)
|
|
assert parse_duration("4h") == timedelta(hours=4)
|
|
assert parse_duration("1d") == timedelta(days=1)
|
|
assert parse_duration("90s") == timedelta(seconds=90)
|
|
|
|
|
|
def test_parse_duration_invalid() -> None:
|
|
with pytest.raises(ValueError, match="Invalid duration"):
|
|
parse_duration("4hours")
|
|
with pytest.raises(ValueError, match="positive"):
|
|
parse_duration("0h")
|
|
|
|
|
|
def test_cap_duration() -> None:
|
|
assert cap_duration("4h", "24h") == "4h"
|
|
assert cap_duration("48h", "24h") == "24h"
|
|
|
|
|
|
def test_resolve_initial_ttl() -> None:
|
|
profile = _profile()
|
|
assert resolve_initial_ttl(profile, None) == "4h"
|
|
assert resolve_initial_ttl(profile, "2h") == "2h"
|
|
assert resolve_initial_ttl(profile, "48h") == "24h"
|
|
|
|
|
|
def test_extend_expires_at_caps_at_max() -> None:
|
|
anchor = datetime(2026, 6, 24, 10, 0, tzinfo=UTC)
|
|
current = anchor + timedelta(hours=23)
|
|
new_expires, applied = extend_expires_at(
|
|
current,
|
|
anchor=anchor,
|
|
extension="4h",
|
|
max_duration="24h",
|
|
)
|
|
assert new_expires == anchor + timedelta(hours=24)
|
|
assert applied == "1h"
|
|
|
|
|
|
def test_extend_expires_at_at_max_raises() -> None:
|
|
anchor = datetime(2026, 6, 24, 10, 0, tzinfo=UTC)
|
|
current = anchor + timedelta(hours=24)
|
|
with pytest.raises(ValueError, match="profile max"):
|
|
extend_expires_at(
|
|
current,
|
|
anchor=anchor,
|
|
extension="1h",
|
|
max_duration="24h",
|
|
)
|
|
|
|
|
|
def test_format_timedelta() -> None:
|
|
assert format_timedelta(timedelta(hours=2)) == "2h"
|
|
assert format_timedelta(timedelta(minutes=30)) == "30m"
|
|
|
|
|
|
def test_is_past_expiry_and_idle() -> None:
|
|
now = datetime(2026, 6, 24, 12, 0, tzinfo=UTC)
|
|
assert is_past_expiry(now - timedelta(minutes=1), now=now)
|
|
assert not is_past_expiry(now + timedelta(minutes=1), now=now)
|
|
updated = now - timedelta(hours=2)
|
|
assert is_idle_expired(updated, "1h", now=now)
|
|
assert not is_idle_expired(updated, "4h", now=now)
|
|
|
|
|
|
@pytest.fixture
|
|
def store(tmp_path: Path) -> SandboxStore:
|
|
return SandboxStore(path=tmp_path / "sandboxes.json")
|
|
|
|
|
|
def test_find_expire_candidates_ttl_and_idle(store: SandboxStore) -> None:
|
|
now = datetime(2026, 6, 24, 12, 0, tzinfo=UTC)
|
|
store.save(
|
|
SandboxStatus(
|
|
sandbox_id="expired1",
|
|
profile_id="profile.compose-e2e",
|
|
extension_id="ext.compose-ssh",
|
|
state=SandboxState.READY,
|
|
consumer=Consumer(actor=ActorType.ADM, project="sand-boxer"),
|
|
expires_at=now - timedelta(minutes=5),
|
|
created_at=now - timedelta(hours=5),
|
|
updated_at=now - timedelta(hours=5),
|
|
ready_at=now - timedelta(hours=5),
|
|
)
|
|
)
|
|
store.save(
|
|
SandboxStatus(
|
|
sandbox_id="idle1",
|
|
profile_id="profile.sandbox-canary",
|
|
extension_id="ext.compose-ssh",
|
|
state=SandboxState.READY,
|
|
consumer=Consumer(actor=ActorType.ADM, project="sand-boxer"),
|
|
expires_at=now + timedelta(hours=2),
|
|
created_at=now - timedelta(hours=5),
|
|
updated_at=now - timedelta(hours=3),
|
|
ready_at=now - timedelta(hours=5),
|
|
)
|
|
)
|
|
|
|
with patch("sandboxer.lifecycle.expire.load_profile") as load_profile:
|
|
load_profile.side_effect = lambda pid: _profile(
|
|
idle_reap="2h" if pid == "profile.sandbox-canary" else None
|
|
)
|
|
candidates = find_expire_candidates(store, now=now)
|
|
|
|
reasons = {c.sandbox_id: c.reason for c in candidates}
|
|
assert reasons["expired1"] == "ttl"
|
|
assert reasons["idle1"] == "idle"
|
|
|
|
|
|
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"}
|
|
|
|
|
|
def test_manager_create_sets_expires_at(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"),
|
|
ttl="2h",
|
|
)
|
|
fake = FakeBackend()
|
|
with (
|
|
patch("sandboxer.core.manager.resolve_backend", return_value=fake),
|
|
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.ttl == "2h"
|
|
assert status.expires_at is not None
|
|
assert status.ready_at is not None
|
|
assert status.expires_at > status.ready_at
|
|
|
|
|
|
def test_manager_extend_ttl(store: SandboxStore) -> None:
|
|
now = datetime.now(UTC)
|
|
store.save(
|
|
SandboxStatus(
|
|
sandbox_id="live1234",
|
|
profile_id="profile.compose-e2e",
|
|
extension_id="ext.compose-ssh",
|
|
state=SandboxState.READY,
|
|
consumer=Consumer(actor=ActorType.ADM, project="sand-boxer"),
|
|
host="coulombcore",
|
|
reachability=Reachability(remote_dir="/tmp/x", host="coulombcore"),
|
|
ttl="4h",
|
|
expires_at=now + timedelta(hours=1),
|
|
created_at=now - timedelta(hours=1),
|
|
updated_at=now,
|
|
ready_at=now - timedelta(hours=1),
|
|
)
|
|
)
|
|
manager = SandboxManager(store=store)
|
|
with patch("sandboxer.core.manager.emit_lifecycle_event", return_value=None):
|
|
extended = manager.extend_ttl("live1234", "2h")
|
|
assert extended.expires_at > now + timedelta(hours=1)
|
|
|
|
|
|
def test_manager_expire_dry_run_and_apply(store: SandboxStore) -> None:
|
|
now = datetime.now(UTC)
|
|
store.save(
|
|
SandboxStatus(
|
|
sandbox_id="gone5678",
|
|
profile_id="profile.compose-e2e",
|
|
extension_id="ext.compose-ssh",
|
|
state=SandboxState.READY,
|
|
consumer=Consumer(actor=ActorType.ADM, project="sand-boxer"),
|
|
host="coulombcore",
|
|
reachability=Reachability(
|
|
remote_dir="/tmp/sandboxer/gone5678",
|
|
compose_project="sbx-e2e-gone5678",
|
|
host="coulombcore",
|
|
),
|
|
inputs={"compose_file": "docker-compose.yml"},
|
|
ttl="1h",
|
|
expires_at=now - timedelta(minutes=1),
|
|
created_at=now - timedelta(hours=2),
|
|
updated_at=now - timedelta(hours=2),
|
|
ready_at=now - timedelta(hours=2),
|
|
)
|
|
)
|
|
manager = SandboxManager(store=store)
|
|
fake = FakeBackend()
|
|
|
|
dry = manager.expire(apply=False, now=now)
|
|
assert len(dry) == 1
|
|
assert dry[0].action == "dry-run"
|
|
assert manager.get("gone5678").state == SandboxState.READY
|
|
|
|
with (
|
|
patch("sandboxer.core.manager.resolve_backend", return_value=fake),
|
|
patch("sandboxer.core.manager.emit_lifecycle_event", return_value=None),
|
|
patch("sandboxer.core.manager.load_extension"),
|
|
patch("sandboxer.core.manager.load_profile"),
|
|
):
|
|
applied = manager.expire(apply=True, now=now)
|
|
|
|
assert applied[0].action == "destroyed"
|
|
assert manager.get("gone5678").state == SandboxState.DESTROYED |