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
No EOL
8.6 KiB
Python
265 lines
No EOL
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 |