119 lines
5.8 KiB
Python
119 lines
5.8 KiB
Python
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from k1link.device_plugins.xgrids_k1 import xgrids_k1_archive_source
|
|
from k1link.sessions import SessionStore
|
|
from k1link.sessions.display_profile import load_display_profile, save_display_profile
|
|
from k1link.web.session_api import build_session_router
|
|
from test_session_api import make_legacy_session
|
|
|
|
|
|
def settings(**patch):
|
|
return dict(projection="3d", point_size=0.5, color_mode="height", palette="viridis",
|
|
custom_color="#112233", accumulation_seconds=600, accumulation_max_seconds=600,
|
|
point_decimation_percent=49.5, show_points=True, show_trajectory=True,
|
|
show_grid=True, show_labels=False, show_camera_frustums=True, **patch)
|
|
|
|
|
|
def test_atomic_session_metadata_keeps_evidence_and_isolates_sessions(tmp_path: Path):
|
|
a, b = tmp_path / "a", tmp_path / "b"
|
|
a.mkdir(); b.mkdir()
|
|
(a / "method.json").write_text('{"immutable":true}')
|
|
assert load_display_profile(a, "a") is None
|
|
save_display_profile(a, "a", settings())
|
|
assert load_display_profile(a, "a")["scene_settings"] == settings()
|
|
assert load_display_profile(b, "b") is None
|
|
assert (a / "method.json").read_text() == '{"immutable":true}'
|
|
assert not list(a.glob(".display-profile-*"))
|
|
with pytest.raises(ValueError):
|
|
load_display_profile(a, "b")
|
|
|
|
|
|
def test_profile_symlink_is_not_followed(tmp_path: Path):
|
|
other = tmp_path / "other.json"
|
|
other.write_text("original")
|
|
(tmp_path / "display-profile.json").symlink_to(other)
|
|
with pytest.raises(ValueError):
|
|
save_display_profile(tmp_path, "a", settings())
|
|
assert other.read_text() == "original"
|
|
|
|
|
|
def test_failed_atomic_replace_keeps_previous_profile(tmp_path: Path, monkeypatch):
|
|
save_display_profile(tmp_path, 'a', settings())
|
|
previous = (tmp_path / 'display-profile.json').read_bytes()
|
|
def fail(*args):
|
|
raise OSError('disk unavailable')
|
|
monkeypatch.setattr('k1link.sessions.display_profile.os.replace', fail)
|
|
changed = settings(); changed['point_decimation_percent'] = 80
|
|
with pytest.raises(OSError): save_display_profile(tmp_path, 'a', changed)
|
|
assert (tmp_path / 'display-profile.json').read_bytes() == previous
|
|
assert not list(tmp_path.glob('.display-profile-*'))
|
|
|
|
|
|
def test_profile_api_roundtrip_strict_validation_and_no_capture_mutation(tmp_path: Path):
|
|
repository = tmp_path / "repo"
|
|
sessions = repository / "sessions"
|
|
session = make_legacy_session(sessions, "20260921T134309Z_viewer_live")
|
|
store = SessionStore(repository, data_dir=tmp_path / "data")
|
|
store.reconcile_archive(xgrids_k1_archive_source(sessions))
|
|
app = FastAPI(); app.include_router(build_session_router(store))
|
|
url = f"/api/v1/observation-sessions/{session.name}/display-profile"
|
|
with TestClient(app) as client:
|
|
assert client.get(url).json() is None
|
|
response = client.put(url, json={"scene_settings": settings()})
|
|
assert response.status_code == 200
|
|
assert client.get(url).json() == response.json()
|
|
assert json.loads((session / "display-profile.json").read_text()) == response.json()
|
|
for value in (-1, 100.1):
|
|
invalid = settings(); invalid["point_decimation_percent"] = value
|
|
assert client.put(url, json={"scene_settings": invalid}).status_code == 422
|
|
for value in (0, 50, 100):
|
|
valid = settings(); valid["point_decimation_percent"] = value
|
|
assert client.put(url, json={"scene_settings": valid}).status_code == 200
|
|
assert client.put(url, json={"scene_settings": settings(), "path": "/tmp"}).status_code == 422
|
|
assert client.get("/api/v1/observation-sessions/missing/display-profile").status_code == 404
|
|
|
|
|
|
def test_point_display_pins_exact_prepared_generation_and_releases_on_error(tmp_path):
|
|
from types import SimpleNamespace
|
|
repository = tmp_path / 'repo'
|
|
session = make_legacy_session(repository / 'sessions', '20260921T134309Z_viewer_live')
|
|
store = SessionStore(repository, data_dir=tmp_path / 'data')
|
|
store.reconcile_archive(xgrids_k1_archive_source(repository / 'sessions'))
|
|
command = store.prepare_replay(session.name)
|
|
path = tmp_path / 'prepared.rrd'
|
|
snapshot = SimpleNamespace(state='ready', preparation_id='prep-1', command=command,
|
|
recording=SimpleNamespace(path=path, sha256='a' * 64))
|
|
releases, calls = [], []
|
|
manager = SimpleNamespace(status=lambda _: snapshot,
|
|
pin_ready=lambda *a, **k: (snapshot, lambda: releases.append(True)))
|
|
fail = False
|
|
def renderer(command, **kwargs):
|
|
calls.append(kwargs)
|
|
assert kwargs['prepared_recording_path'] == path
|
|
assert 'source_generation' not in kwargs
|
|
if fail:
|
|
raise ValueError('bad prepared source')
|
|
yield b'NPD1'
|
|
yield b'\0' * 4
|
|
app = FastAPI()
|
|
app.include_router(build_session_router(store, recording_preparation_manager=manager,
|
|
point_display_renderers={command.plugin_id: renderer}))
|
|
url = f'/api/v1/observation-sessions/{session.name}/point-display.rrd'
|
|
request = dict(application_id='nodedc_mission_core_recorded', recording_id='test-recording',
|
|
color_mode='intensity', palette='turbo', custom_color='#112233',
|
|
point_decimation_percent=86.2, display_bank='b' * 32, source_generation='a' * 64)
|
|
with TestClient(app) as client:
|
|
assert client.post(url, json={**request, 'source_generation': 'f' * 64}).status_code == 412
|
|
assert not calls and not releases
|
|
result = client.post(url, json=request)
|
|
assert result.status_code == 200 and result.content == b'NPD1' + b'\0' * 4
|
|
assert len(releases) == 1
|
|
fail = True
|
|
assert client.post(url, json=request).status_code == 409
|
|
assert len(releases) == 2
|