import hashlib import json import numpy as np import pytest import rerun as rr from rerun.experimental import RrdReader from k1link.sessions.overview_comparison import load_comparison, publish_comparison from k1link.sessions.overview_spatial import render_spatial_update, spatial_metadata @pytest.fixture def pair(tmp_path): root = tmp_path / "previews" inputs = dict( session_id="recording", overview_generation="a" * 64, source_digests={"raw": "b" * 64}, map_generation="c" * 64, source_points=30, original_path_m=12.0, corrected_path_m=11.9, original=[[0, 0, 0], [1, 2, 3], [4, 5, 8]], corrected=[[0, 0, 0], [1, 2, 2], [4, 5, 7]], original_route=[[0, 0, 0], [4, 5, 8]], corrected_route=[[0, 0, 0], [4, 5, 7]], ) generation = publish_comparison(root, **inputs) return root, generation, inputs def load(pair, **changes): root, generation, inputs = pair values = {k: inputs[k] for k in ("overview_generation", "session_id", "source_digests")} return load_comparison(root, **{**values, **changes}, generation=generation) def test_publication_is_source_bound_idempotent_and_not_a_new_session(pair): root, generation, inputs = pair assert publish_comparison(root, **inputs) == generation preview = load(pair) np.testing.assert_array_equal(preview.original, inputs["original"]) assert preview.document["view_only"] is True assert sorted(p.name for p in root.iterdir()) == [generation, "selected"] assert ( load_comparison( root, inputs["overview_generation"], "recording", inputs["source_digests"] ).generation == generation ) assert load_comparison(root, "d" * 64, "recording", inputs["source_digests"]) is None @pytest.mark.parametrize( "changes", [ dict(session_id="other"), dict(overview_generation="d" * 64), dict(source_digests={"raw": "e" * 64}), ], ) def test_does_not_cross_source_boundaries(pair, changes): with pytest.raises(ValueError, match="another recording"): load(pair, **changes) @pytest.mark.parametrize("filename", ["manifest.json", "geometry.npz"]) def test_corrupt_artifacts_fail_closed(pair, filename): root, generation, _ = pair path = root / generation / filename path.write_bytes(path.read_bytes() + b"X") with pytest.raises(ValueError, match="changed"): load(pair) def test_geometry_rejects_non_finite_or_mismatched_pairs(pair): root, _, inputs = pair with pytest.raises(ValueError, match="geometry"): publish_comparison(root, **{**inputs, "corrected": [[float("nan"), 0, 0]]}) def test_path_traversal_and_symlinks_are_rejected(pair): root, generation, inputs = pair with pytest.raises(ValueError, match="identity"): load_comparison( root, inputs["overview_generation"], "recording", inputs["source_digests"], "../source" ) path = root / generation / "geometry.npz" path.rename(path.with_suffix(".original")) path.symlink_to(path.with_suffix(".original")) with pytest.raises(ValueError, match="artifact"): load(pair) def test_switch_replaces_cloud_route_endpoints_without_camera_blueprint(pair, tmp_path): preview = load(pair) original = tmp_path / "scene.rrd" recording = rr.RecordingStream("overview-test") recording.save(original) recording.log("world/cloud", rr.Points3D(preview.original, colors=[30, 40, 50]), static=True) recording.flush() recording.disconnect() digest = hashlib.sha256(original.read_bytes()).hexdigest() metadata = spatial_metadata(original, preview) assert metadata["comparison"]["height_max_m"] == 80 assert metadata["comparison"]["source_points"] == 30 colors = [] for representation in ("original", "corrected", "original"): data, count, eye = render_spatial_update( original, None, None, comparison=preview, representation=representation ) assert count == 3 and eye is None path = tmp_path / f"{representation}.rrd" path.write_bytes(data) reader = RrdReader(path) assert len(reader.recordings()) == 1 assert ( reader.recordings()[0].recording_id == RrdReader(original).recordings()[0].recording_id ) chunks = {c.entity_path: c.to_record_batch() for c in reader.stream()} assert set(chunks) == {"/world/cloud", "/world/route", "/world/endpoints"} points = ( chunks["/world/cloud"] .column("Points3D:positions")[0] .values.values.to_numpy() .reshape(-1, 3) ) expected = getattr(preview, representation) np.testing.assert_array_equal(points, expected) endpoints = ( chunks["/world/endpoints"] .column("Points3D:positions")[0] .values.values.to_numpy() .reshape(-1, 3) ) np.testing.assert_array_equal( endpoints, getattr(preview, representation + "_route")[[0, -1]] ) colors.append(chunks["/world/cloud"].column("Points3D:colors")[0].values.to_numpy()) np.testing.assert_array_equal(colors[0], colors[1]) assert hashlib.sha256(original.read_bytes()).hexdigest() == digest assert ( render_spatial_update(original, 2.5, None, comparison=preview, representation="corrected")[ 1 ] == 2 ) assert ( render_spatial_update(original, 2.5, None, comparison=preview, representation="original")[1] == 1 ) eyes = [ render_spatial_update(original, None, "top", comparison=preview, representation=x)[2] for x in ("original", "corrected") ] assert eyes[0] == eyes[1] with pytest.raises(ValueError, match="requires a pinned"): render_spatial_update(original, None, None, representation="corrected") def test_pinned_comparison_does_not_follow_new_selection(pair): root, generation, inputs = pair newer = publish_comparison(root, **{**inputs, "corrected": [[0, 0, 0], [1, 2, 1], [4, 5, 6]]}) assert generation != newer assert load(pair).generation == generation assert ( json.loads((root / "selected" / (inputs["overview_generation"] + ".json")).read_text())[ "generation" ] == newer ) def test_http_requires_pinned_version_and_never_falls_back_to_original(pair, tmp_path): from types import SimpleNamespace from fastapi import FastAPI from fastapi.testclient import TestClient from k1link.web.session_overview_api import build_session_overview_router path = tmp_path / "scene.rrd" recording = rr.RecordingStream("http-overview-test") recording.save(path) recording.log("world/cloud", rr.Points3D([[0, 0, 0]], colors=[10, 20, 30]), static=True) recording.flush() recording.disconnect() root, generation, inputs = pair def scene(session, overview): if session != "recording" or overview != inputs["overview_generation"]: raise ValueError("source changed") return path def comparison(session, overview, pinned=None): scene(session, overview) return load_comparison(root, overview, session, inputs["source_digests"], pinned) app = FastAPI() app.include_router( build_session_overview_router(SimpleNamespace(scene=scene, comparison=comparison, default_representation=lambda *_: "original")) ) with TestClient(app) as client: url = "/api/v1/observation-sessions/recording/overview/spatial" response = client.get(url, params={"generation": inputs["overview_generation"]}) assert response.status_code == 200 assert response.json()["comparison"]["generation"] == generation request = dict(generation=inputs["overview_generation"], representation="corrected") assert client.post(url, json=request).status_code == 409 assert ( client.post(url, json={**request, "comparison_generation": "f" * 64}).status_code == 409 ) assert ( client.post(url, json={**request, "comparison_generation": "../source"}).status_code == 422 ) response = client.post(url, json={**request, "comparison_generation": generation}) assert response.status_code == 200 assert "X-Overview-Eye" not in response.headers assert response.headers["X-Overview-Visible-Points"] == "3"