115 lines
4.9 KiB
Python
115 lines
4.9 KiB
Python
"""Executable completeness contract for CellManager and its CLI."""
|
|
from __future__ import annotations
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
import pytest
|
|
from aios_v2r1.cell_manager import CellManager
|
|
from aios_v2r1.cli import main
|
|
from aios_v2r1.tenant_context import ValidationError
|
|
|
|
def test_one_cell_per_tenant(tmp_path: Path) -> None:
|
|
manager = CellManager(tmp_path / "cells.json")
|
|
manager.create("tenant-a", "cell-a", "v1", "digest-a")
|
|
with pytest.raises(ValidationError, match="already"):
|
|
manager.create("tenant-a", "cell-b", "v1", "digest-b")
|
|
|
|
def test_explicit_state_graph_accepts_the_recovery_cycle(tmp_path: Path) -> None:
|
|
manager = CellManager(tmp_path / "cells.json")
|
|
manager.create("tenant-a", "cell-a", "v1", "digest-a")
|
|
for state in ("starting", "healthy", "upgrading", "rollback", "healthy"):
|
|
assert manager.transition("tenant-a", state).state == state
|
|
|
|
def test_explicit_state_graph_rejects_jump_from_created_to_healthy(tmp_path: Path) -> None:
|
|
manager = CellManager(tmp_path / "cells.json")
|
|
manager.create("tenant-a", "cell-a", "v1", "digest-a")
|
|
with pytest.raises(ValidationError, match="transition"):
|
|
manager.transition("tenant-a", "healthy")
|
|
|
|
def test_persistence_uses_atomic_replace(tmp_path: Path) -> None:
|
|
manager = CellManager(tmp_path / "cells.json")
|
|
with patch("aios_v2r1.cell_manager.os.replace", wraps=os.replace) as replaced:
|
|
manager.create("tenant-a", "cell-a", "v1", "digest-a")
|
|
assert replaced.called
|
|
|
|
def test_concurrent_distinct_tenant_creates_are_not_lost(tmp_path: Path) -> None:
|
|
registry = tmp_path / "cells.json"
|
|
manager = CellManager(registry)
|
|
def create(index: int) -> None:
|
|
manager.create(f"tenant-{index}", f"cell-{index}", "v1", f"digest-{index}")
|
|
with ThreadPoolExecutor(max_workers=16) as executor:
|
|
list(executor.map(create, range(24)))
|
|
data = json.loads(registry.read_text(encoding="utf-8"))
|
|
assert len(data["cells"]) == 24
|
|
|
|
def test_concurrent_same_tenant_has_exactly_one_winner(tmp_path: Path) -> None:
|
|
manager = CellManager(tmp_path / "cells.json")
|
|
def create(index: int) -> str:
|
|
try:
|
|
manager.create("tenant-a", f"cell-{index}", "v1", f"digest-{index}")
|
|
return "created"
|
|
except ValidationError:
|
|
return "rejected"
|
|
with ThreadPoolExecutor(max_workers=16) as executor:
|
|
results = list(executor.map(create, range(16)))
|
|
assert results.count("created") == 1
|
|
assert results.count("rejected") == 15
|
|
|
|
def test_corrupt_registry_fails_closed_with_domain_error(tmp_path: Path) -> None:
|
|
registry = tmp_path / "cells.json"
|
|
manager = CellManager(registry)
|
|
registry.write_text('{"cells":{"tenant-a":', encoding="utf-8")
|
|
with pytest.raises(ValidationError, match="registry unavailable") as raised:
|
|
manager.status("tenant-a")
|
|
assert "JSON" not in type(raised.value).__name__
|
|
assert raised.value.__cause__ is None
|
|
assert raised.value.__context__ is None
|
|
|
|
def test_restart_preserves_cell_status(tmp_path: Path) -> None:
|
|
registry = tmp_path / "cells.json"
|
|
CellManager(registry).create("tenant-a", "cell-a", "v1", "digest-a")
|
|
assert CellManager(registry).status("tenant-a").cell_id == "cell-a"
|
|
|
|
def test_cli_create_returns_stable_json(
|
|
tmp_path: Path, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
registry = tmp_path / "cells.json"
|
|
assert main([str(registry), "create", "tenant-a", "cell-a", "v1", "digest-a"]) == 0
|
|
output = json.loads(capsys.readouterr().out)
|
|
assert output["tenant_id"] == "tenant-a"
|
|
assert output["state"] == "created"
|
|
|
|
def test_cli_domain_failure_returns_code_3_and_json(
|
|
tmp_path: Path, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
registry = tmp_path / "cells.json"
|
|
assert main([str(registry), "status", "missing-tenant"]) == 3
|
|
captured = capsys.readouterr()
|
|
assert captured.out == ""
|
|
assert json.loads(captured.err) == {
|
|
"code": "VALIDATION_ERROR", "error": "unknown tenant"}
|
|
|
|
def test_cli_corrupt_registry_is_sanitized(
|
|
tmp_path: Path, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
registry = tmp_path / "cells.json"
|
|
CellManager(registry)
|
|
registry.write_text("TOP-SECRET invalid json", encoding="utf-8")
|
|
assert main([str(registry), "status", "tenant-a"]) == 3
|
|
error = json.loads(capsys.readouterr().err)
|
|
assert error["code"] == "VALIDATION_ERROR"
|
|
assert "TOP-SECRET" not in error["error"]
|
|
|
|
@pytest.mark.parametrize("args", [
|
|
("Tenant-A", "cell-a", "v1", "digest-a"),
|
|
("tenant-a", "../cell", "v1", "digest-a"),
|
|
("tenant-a", "cell-a", "../v1", "digest-a"),
|
|
("tenant-a", "cell-a", "v1", "../digest")])
|
|
def test_create_validates_all_identifiers(
|
|
tmp_path: Path, args: tuple[str, str, str, str]
|
|
) -> None:
|
|
with pytest.raises(ValidationError):
|
|
CellManager(tmp_path / "cells.json").create(*args)
|