499 lines
17 KiB
Python
499 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
from collections.abc import Callable, Iterable, Mapping
|
|
from contextlib import AbstractContextManager
|
|
from pathlib import Path
|
|
from threading import Event, Thread
|
|
from typing import cast
|
|
|
|
import pytest
|
|
import uvicorn
|
|
from fastapi.testclient import TestClient
|
|
from starlette.types import ASGIApp
|
|
|
|
import game_data_capture.runtime as runtime_module
|
|
from game_data_capture.config import AppConfig, CaptureLimits, LifecycleConfig, ServerConfig
|
|
from game_data_capture.ingest.clipboard import IntakeReceipt
|
|
from game_data_capture.ingest.spool_helper import PNG_SIGNATURE
|
|
from game_data_capture.paths import RuntimePaths
|
|
from game_data_capture.runtime import (
|
|
CaptureRuntime,
|
|
InstanceLock,
|
|
IntakePort,
|
|
RuntimeFailure,
|
|
_CaptureControl,
|
|
)
|
|
from game_data_capture.storage import CaptureCorpus, CaptureRepository, StorageDatabase
|
|
from game_data_capture.web.app import WebConfig
|
|
|
|
_PNG = PNG_SIGNATURE + b"unit-png"
|
|
|
|
|
|
def _paths(tmp_path: Path) -> RuntimePaths:
|
|
root = tmp_path / "xdg"
|
|
return RuntimePaths(
|
|
config_dir=(root / "config").resolve(),
|
|
data_dir=(root / "data").resolve(),
|
|
cache_dir=(root / "cache").resolve(),
|
|
runtime_dir=(root / "runtime").resolve(),
|
|
)
|
|
|
|
|
|
def _runtime(tmp_path: Path, *, limits: CaptureLimits | None = None) -> CaptureRuntime:
|
|
return CaptureRuntime(
|
|
AppConfig(
|
|
lifecycle=LifecycleConfig(lock_timeout_seconds=0), limits=limits or CaptureLimits()
|
|
),
|
|
_paths(tmp_path),
|
|
migrations_dir=(Path(__file__).parents[3] / "migrations").resolve(),
|
|
)
|
|
|
|
|
|
def test_one_shot_persists_distinct_occurrences_and_one_blob(tmp_path: Path) -> None:
|
|
runtime = _runtime(tmp_path)
|
|
image = tmp_path / "image.png"
|
|
image.write_bytes(_PNG)
|
|
|
|
first = runtime.ingest_image(image)
|
|
second = runtime.ingest_image(image)
|
|
database = StorageDatabase(runtime.paths.data_dir, runtime.migrations_dir)
|
|
|
|
with database.connect() as connection:
|
|
assert first != second
|
|
assert connection.execute("SELECT count(*) FROM captures").fetchone()[0] == 2
|
|
assert connection.execute("SELECT count(*) FROM blobs").fetchone()[0] == 1
|
|
|
|
|
|
@pytest.mark.parametrize("kind", ["malformed", "oversize", "symlink", "broad-mode"])
|
|
def test_one_shot_rejects_unsafe_png_inputs(tmp_path: Path, kind: str) -> None:
|
|
runtime = _runtime(tmp_path, limits=CaptureLimits(max_image_bytes=len(_PNG)))
|
|
image = tmp_path / "image.png"
|
|
image.write_bytes(_PNG)
|
|
if kind == "malformed":
|
|
image.write_bytes(b"not-a-png")
|
|
elif kind == "oversize":
|
|
image.write_bytes(_PNG + b"x")
|
|
elif kind == "symlink":
|
|
linked = tmp_path / "linked.png"
|
|
linked.symlink_to(image)
|
|
image = linked
|
|
else:
|
|
image.chmod(0o666)
|
|
try:
|
|
with pytest.raises(RuntimeFailure, match="safe PNG"):
|
|
runtime.ingest_image(image)
|
|
finally:
|
|
if image.exists() and not image.is_symlink():
|
|
image.chmod(0o600)
|
|
|
|
|
|
def test_stale_lock_file_is_reused_but_an_active_lock_is_not(tmp_path: Path) -> None:
|
|
paths = _paths(tmp_path)
|
|
paths.runtime_dir.mkdir(parents=True)
|
|
paths.instance_lock_file.write_text("stale", encoding="utf-8")
|
|
paths.instance_lock_file.chmod(0o600)
|
|
first = InstanceLock.acquire(paths.instance_lock_file, 0)
|
|
try:
|
|
with pytest.raises(RuntimeFailure, match="another capture instance is active"):
|
|
InstanceLock.acquire(paths.instance_lock_file, 0)
|
|
finally:
|
|
first.release()
|
|
second = InstanceLock.acquire(paths.instance_lock_file, 0)
|
|
second.release()
|
|
assert paths.instance_lock_file.exists()
|
|
|
|
|
|
class InterruptingIntake:
|
|
def __init__(self) -> None:
|
|
self.shutdown_called = False
|
|
|
|
def reconcile(self) -> list[str]:
|
|
return []
|
|
|
|
def arm(self) -> None:
|
|
return None
|
|
|
|
def tick(self) -> None:
|
|
raise KeyboardInterrupt
|
|
|
|
def take_helper_receipt(self) -> dict[str, object] | None:
|
|
return None
|
|
|
|
def accept_helper_receipt(
|
|
self, payload: Mapping[str, object], *, occurrence_id: str, processing_version: str
|
|
) -> bool:
|
|
return False
|
|
|
|
def take(self) -> IntakeReceipt | None:
|
|
return None
|
|
|
|
def open_verified(self, receipt: IntakeReceipt) -> AbstractContextManager[Iterable[bytes]]:
|
|
raise AssertionError("no receipt should be opened")
|
|
|
|
def acknowledge(self, receipt: IntakeReceipt) -> bool:
|
|
raise AssertionError("no receipt should be acknowledged")
|
|
|
|
def shutdown(self) -> None:
|
|
self.shutdown_called = True
|
|
|
|
|
|
def test_keyboard_interrupt_reaps_intake_and_releases_runtime_lock(tmp_path: Path) -> None:
|
|
intake = InterruptingIntake()
|
|
runtime = CaptureRuntime(
|
|
AppConfig(lifecycle=LifecycleConfig(lock_timeout_seconds=0)),
|
|
_paths(tmp_path),
|
|
migrations_dir=(Path(__file__).parents[3] / "migrations").resolve(),
|
|
intake_factory=lambda _config: intake,
|
|
)
|
|
runtime.watch(poll_seconds=0.001)
|
|
|
|
assert intake.shutdown_called
|
|
reusable = InstanceLock.acquire(runtime.paths.instance_lock_file, 0)
|
|
reusable.release()
|
|
|
|
|
|
def test_cancelled_watch_cleans_up_without_entering_the_tick_loop(tmp_path: Path) -> None:
|
|
cancelled = Event()
|
|
cancelled.set()
|
|
intake = InterruptingIntake()
|
|
runtime = CaptureRuntime(
|
|
AppConfig(lifecycle=LifecycleConfig(lock_timeout_seconds=0)),
|
|
_paths(tmp_path),
|
|
migrations_dir=(Path(__file__).parents[3] / "migrations").resolve(),
|
|
intake_factory=lambda _config: intake,
|
|
)
|
|
runtime.watch(cancelled=cancelled, poll_seconds=0.001)
|
|
assert intake.shutdown_called
|
|
|
|
|
|
def test_one_shot_never_connects_outside_the_utility_data_root(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
runtime = _runtime(tmp_path)
|
|
image = tmp_path / "image.png"
|
|
image.write_bytes(_PNG)
|
|
real_connect = cast(Callable[..., sqlite3.Connection], sqlite3.connect)
|
|
connected: list[Path] = []
|
|
|
|
def checked_connect(database: str | Path, **kwargs: object) -> sqlite3.Connection:
|
|
connected.append(Path(database))
|
|
return real_connect(database, **kwargs)
|
|
|
|
monkeypatch.setattr(sqlite3, "connect", checked_connect)
|
|
runtime.ingest_image(image)
|
|
assert connected
|
|
assert all(path.parent == runtime.paths.data_dir for path in connected)
|
|
|
|
|
|
class ControlledIntake:
|
|
def __init__(self, *, fail_tick: bool = False) -> None:
|
|
self.armed = False
|
|
self.reconciled = False
|
|
self.shutdown_called = Event()
|
|
self.queue_depth = 0
|
|
self.helper_queue_depth = 0
|
|
self.fail_tick = fail_tick
|
|
|
|
def reconcile(self) -> list[str]:
|
|
self.reconciled = True
|
|
return []
|
|
|
|
def arm(self) -> None:
|
|
self.armed = True
|
|
|
|
def tick(self) -> None:
|
|
if self.fail_tick:
|
|
raise OSError("helper fault")
|
|
|
|
def take_helper_receipt(self) -> dict[str, object] | None:
|
|
return None
|
|
|
|
def accept_helper_receipt(
|
|
self, payload: Mapping[str, object], *, occurrence_id: str, processing_version: str
|
|
) -> bool:
|
|
return False
|
|
|
|
def take(self) -> IntakeReceipt | None:
|
|
return None
|
|
|
|
def open_verified(self, receipt: IntakeReceipt) -> AbstractContextManager[Iterable[bytes]]:
|
|
raise AssertionError("no receipt should be opened")
|
|
|
|
def acknowledge(self, receipt: IntakeReceipt) -> bool:
|
|
raise AssertionError("no receipt should be acknowledged")
|
|
|
|
def shutdown(self) -> None:
|
|
self.shutdown_called.set()
|
|
|
|
|
|
class FaultingCleanupIntake(ControlledIntake):
|
|
def __init__(self) -> None:
|
|
super().__init__(fail_tick=True)
|
|
self.cleanup_started = Event()
|
|
self.cleanup_release = Event()
|
|
self._shutdown_calls = 0
|
|
|
|
def shutdown(self) -> None:
|
|
self._shutdown_calls += 1
|
|
self.cleanup_started.set()
|
|
if self._shutdown_calls == 1:
|
|
self.cleanup_release.wait()
|
|
self.shutdown_called.set()
|
|
|
|
|
|
class MalformedQueueIntake:
|
|
@property
|
|
def queue_depth(self) -> object:
|
|
raise OSError("private intake failure")
|
|
|
|
@property
|
|
def helper_queue_depth(self) -> object:
|
|
return True
|
|
|
|
|
|
class EmptyRepository:
|
|
def receive(
|
|
self,
|
|
capture_id: str,
|
|
receipt_token: str,
|
|
chunks: Iterable[bytes],
|
|
*,
|
|
media_type: str,
|
|
max_bytes: int,
|
|
) -> str:
|
|
raise AssertionError("no receipt should be stored")
|
|
|
|
|
|
def test_control_serializes_repeated_start_stop_and_reports_bounded_queue(tmp_path: Path) -> None:
|
|
intake = ControlledIntake()
|
|
runtime = CaptureRuntime(
|
|
AppConfig(lifecycle=LifecycleConfig(lock_timeout_seconds=0)),
|
|
_paths(tmp_path),
|
|
migrations_dir=(Path(__file__).parents[3] / "migrations").resolve(),
|
|
intake_factory=lambda _config: intake,
|
|
)
|
|
control = _CaptureControl(runtime, cast(CaptureRepository, EmptyRepository()))
|
|
outcomes: list[str] = []
|
|
|
|
def start() -> None:
|
|
outcomes.append(control.start_session())
|
|
|
|
first, second = Thread(target=start), Thread(target=start)
|
|
first.start()
|
|
second.start()
|
|
first.join()
|
|
second.join()
|
|
|
|
assert sorted(outcomes) == ["already_armed", "started"]
|
|
assert intake.reconciled and intake.armed
|
|
assert control.status().queue_depth == 0
|
|
assert control.status().queue_capacity == runtime.config.limits.max_queue_items
|
|
assert control.stop_session() in {"stopped", "already_stopped"}
|
|
assert control.stop_session() == "already_stopped"
|
|
assert intake.shutdown_called.wait(0.1)
|
|
|
|
|
|
def test_control_status_combines_and_bounds_both_intake_queues(tmp_path: Path) -> None:
|
|
intake = ControlledIntake()
|
|
runtime = CaptureRuntime(
|
|
AppConfig(lifecycle=LifecycleConfig(lock_timeout_seconds=0)),
|
|
_paths(tmp_path),
|
|
migrations_dir=(Path(__file__).parents[3] / "migrations").resolve(),
|
|
intake_factory=lambda _config: intake,
|
|
)
|
|
control = _CaptureControl(runtime, cast(CaptureRepository, EmptyRepository()))
|
|
|
|
intake.queue_depth, intake.helper_queue_depth = 2, 3
|
|
control._intake = cast(IntakePort, intake)
|
|
assert control.status().queue_depth == 5
|
|
intake.queue_depth, intake.helper_queue_depth = 1000, -1
|
|
assert control.status().queue_depth == runtime.config.limits.max_queue_items
|
|
control._intake = cast(IntakePort, MalformedQueueIntake())
|
|
assert control.status().queue_depth == 0
|
|
|
|
|
|
def test_control_turns_background_intake_fault_into_bounded_disarmed_state(tmp_path: Path) -> None:
|
|
intake = ControlledIntake(fail_tick=True)
|
|
runtime = CaptureRuntime(
|
|
AppConfig(lifecycle=LifecycleConfig(lock_timeout_seconds=0)),
|
|
_paths(tmp_path),
|
|
migrations_dir=(Path(__file__).parents[3] / "migrations").resolve(),
|
|
intake_factory=lambda _config: intake,
|
|
)
|
|
control = _CaptureControl(runtime, cast(CaptureRepository, EmptyRepository()))
|
|
|
|
assert control.start_session() == "started"
|
|
assert intake.shutdown_called.wait(0.2)
|
|
assert control.stop_session() in {"stopped", "already_stopped"}
|
|
assert control.status().armed is False
|
|
assert control.status().queue_depth == 0
|
|
|
|
|
|
def test_service_retains_its_lock_until_faulting_worker_cleanup_finishes(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
intake = FaultingCleanupIntake()
|
|
runtime = CaptureRuntime(
|
|
AppConfig(lifecycle=LifecycleConfig(lock_timeout_seconds=0, shutdown_timeout_seconds=1)),
|
|
_paths(tmp_path),
|
|
migrations_dir=(Path(__file__).parents[3] / "migrations").resolve(),
|
|
intake_factory=lambda _config: intake,
|
|
)
|
|
captured: dict[str, _CaptureControl] = {}
|
|
|
|
def factory(**kwargs: object) -> object:
|
|
captured["control"] = cast(_CaptureControl, kwargs["capture_control"])
|
|
return object()
|
|
|
|
def runner(_: object, **__: object) -> None:
|
|
assert captured["control"].start_session() == "started"
|
|
assert intake.cleanup_started.wait(0.2)
|
|
|
|
monkeypatch.setattr(runtime_module, "create_app", factory)
|
|
monkeypatch.setattr(uvicorn, "run", runner)
|
|
failure: list[RuntimeFailure] = []
|
|
|
|
def serve() -> None:
|
|
try:
|
|
runtime.serve()
|
|
except RuntimeFailure as error:
|
|
failure.append(error)
|
|
|
|
service = Thread(target=serve)
|
|
service.start()
|
|
try:
|
|
service.join(timeout=2)
|
|
assert not service.is_alive()
|
|
assert failure
|
|
with pytest.raises(RuntimeFailure, match="another capture instance"):
|
|
InstanceLock.acquire(runtime.paths.instance_lock_file, 0)
|
|
outcomes: list[str] = []
|
|
start = Thread(target=lambda: outcomes.append(captured["control"].start_session()))
|
|
stop = Thread(target=lambda: outcomes.append(captured["control"].stop_session()))
|
|
start.start()
|
|
stop.start()
|
|
start.join()
|
|
stop.join()
|
|
assert sorted(outcomes) == ["conflict", "conflict"]
|
|
finally:
|
|
intake.cleanup_release.set()
|
|
for _ in range(50):
|
|
try:
|
|
reusable = InstanceLock.acquire(runtime.paths.instance_lock_file, 0)
|
|
except RuntimeFailure:
|
|
Event().wait(0.01)
|
|
else:
|
|
reusable.release()
|
|
break
|
|
else:
|
|
pytest.fail("service lock was not released after worker cleanup")
|
|
|
|
|
|
def test_serve_constructs_durable_corpus_without_auto_arming_and_uses_one_fragment_token(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
created: list[ControlledIntake] = []
|
|
token = "A" * 43
|
|
|
|
def factory(_: object) -> ControlledIntake:
|
|
intake = ControlledIntake()
|
|
created.append(intake)
|
|
return intake
|
|
|
|
runtime = CaptureRuntime(
|
|
AppConfig(
|
|
server=ServerConfig(host="127.0.0.1", port=9876),
|
|
lifecycle=LifecycleConfig(lock_timeout_seconds=0),
|
|
),
|
|
_paths(tmp_path),
|
|
migrations_dir=(Path(__file__).parents[3] / "migrations").resolve(),
|
|
intake_factory=factory,
|
|
)
|
|
monkeypatch.setattr(runtime_module, "new_bootstrap_token", lambda: token)
|
|
received: dict[str, object] = {}
|
|
|
|
def runner(app: object, **kwargs: object) -> None:
|
|
received.update(kwargs)
|
|
with TestClient(cast(ASGIApp, app), base_url="http://127.0.0.1:9876") as client:
|
|
assert client.get("/health").json() == {"status": "ok"}
|
|
assert client.get("/").status_code == 200
|
|
assert created == []
|
|
bootstrapped = client.post(
|
|
"/api/bootstrap",
|
|
json={"token": token},
|
|
headers={"Origin": "http://127.0.0.1:9876"},
|
|
)
|
|
csrf = bootstrapped.json()["csrf_token"]
|
|
assert client.get("/api/session").json()["armed"] is False
|
|
started = client.post(
|
|
"/api/session/start",
|
|
json={"confirmation": "START"},
|
|
headers={"Origin": "http://127.0.0.1:9876", "X-CSRF-Token": csrf},
|
|
)
|
|
assert started.json()["result"] == "started"
|
|
assert created[0].reconciled and created[0].armed
|
|
|
|
monkeypatch.setattr(uvicorn, "run", runner)
|
|
runtime.serve()
|
|
|
|
output = capsys.readouterr().out
|
|
assert output.count(token) == 1
|
|
assert f"http://127.0.0.1:9876/#bootstrap={token}" in output
|
|
assert received == {
|
|
"host": "127.0.0.1",
|
|
"port": 9876,
|
|
"reload": False,
|
|
"workers": 1,
|
|
"proxy_headers": False,
|
|
}
|
|
assert created[0].shutdown_called.wait(0.1)
|
|
reusable = InstanceLock.acquire(runtime.paths.instance_lock_file, 0)
|
|
reusable.release()
|
|
|
|
|
|
def test_serve_loads_bounded_corpus_history_independently_of_intake_capacity(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
runtime = CaptureRuntime(
|
|
AppConfig(
|
|
limits=CaptureLimits(max_queue_items=64),
|
|
lifecycle=LifecycleConfig(lock_timeout_seconds=0),
|
|
),
|
|
_paths(tmp_path),
|
|
migrations_dir=(Path(__file__).parents[3] / "migrations").resolve(),
|
|
)
|
|
captured: dict[str, object] = {}
|
|
|
|
def factory(**kwargs: object) -> object:
|
|
captured.update(kwargs)
|
|
return object()
|
|
|
|
def runner(_: object, **__: object) -> None:
|
|
control = cast(_CaptureControl, captured["capture_control"])
|
|
assert control.status().queue_capacity == runtime.config.limits.max_queue_items
|
|
|
|
monkeypatch.setattr(runtime_module, "create_app", factory)
|
|
monkeypatch.setattr(uvicorn, "run", runner)
|
|
|
|
runtime.serve()
|
|
|
|
web_config = cast(WebConfig, captured["config"])
|
|
assert web_config.max_capture_items == 256
|
|
assert web_config.max_capture_items > runtime.config.limits.max_queue_items
|
|
|
|
|
|
def test_service_startup_reconciliation_failure_releases_the_instance_lock(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
runtime = _runtime(tmp_path)
|
|
|
|
def failed_reconciliation(self: CaptureCorpus, *, limit: int = 64) -> int:
|
|
raise OSError("private deletion recovery fault")
|
|
|
|
monkeypatch.setattr(CaptureCorpus, "reconcile_deletions", failed_reconciliation)
|
|
with pytest.raises(RuntimeFailure, match="initialization failed"):
|
|
runtime.serve()
|
|
reusable = InstanceLock.acquire(runtime.paths.instance_lock_file, 0)
|
|
reusable.release()
|