223 lines
8.4 KiB
Python
223 lines
8.4 KiB
Python
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"
|