261 lines
9.7 KiB
Python
261 lines
9.7 KiB
Python
import json
|
|
import threading
|
|
from types import SimpleNamespace
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from k1link.missions.drafts import MissionDrafts
|
|
from k1link.missions.versioned_sources import VersionedPlanningSources
|
|
from k1link.reconstruction.map_version import MapVersion, publish_map_version, sha256
|
|
|
|
|
|
@pytest.fixture
|
|
def fixture(tmp_path):
|
|
p = np.array([[0.0, 0, 0], [10.0, 0, 0], [20.0, 0, 0]])
|
|
corrected = p + np.array([[0.0, 0, 0], [0, 0, 1], [0, 0, 2]])
|
|
frames = np.array([[0, 10, 0, 2], [1, 11, 2, 2], [2, 12, 4, 2]], float)
|
|
points = np.concatenate([xyz + [[0, 0, 0], [0, 0, 10]] for xyz in corrected]).astype("<f4")
|
|
points.tofile(tmp_path / "points.f32")
|
|
np.savez(
|
|
tmp_path / "trajectory.npz",
|
|
positions=corrected,
|
|
orientations_xyzw=np.tile([0.0, 0, 0, 1], (3, 1)),
|
|
receipt_time_s=np.arange(3.0),
|
|
source_distance_m=np.array([0.0, 10, 20]),
|
|
frame_source_distance_m=np.array([0.0, 10, 20]),
|
|
frames=frames,
|
|
)
|
|
source = dict(
|
|
schema_version="missioncore.planning-source/v1",
|
|
session_id="source-a",
|
|
generation="a" * 64,
|
|
label="Original",
|
|
units="m",
|
|
frame_id="session/source-a",
|
|
source_digests={"primary": "b" * 64},
|
|
decode_errors=0,
|
|
path_m=20.0,
|
|
poses=[
|
|
dict(
|
|
index=i,
|
|
message_index=i * 2 + 1,
|
|
position=xyz.tolist(),
|
|
elapsed_s=i,
|
|
distance_m=float(i * 10),
|
|
)
|
|
for i, xyz in enumerate(p)
|
|
],
|
|
)
|
|
(tmp_path / "review.json").write_text('{"kind":"synthetic-test-only"}')
|
|
|
|
def publish(**overrides):
|
|
arguments = dict(
|
|
root=tmp_path / "versions",
|
|
source=source,
|
|
points=tmp_path / "points.f32",
|
|
trajectory=tmp_path / "trajectory.npz",
|
|
expected_points_sha256=sha256(tmp_path / "points.f32"),
|
|
expected_trajectory_sha256=sha256(tmp_path / "trajectory.npz"),
|
|
evidence={"review.json": (tmp_path / "review.json", sha256(tmp_path / "review.json"))},
|
|
method={"algorithm": "synthetic-fixture/v1"},
|
|
label="Corrected candidate",
|
|
)
|
|
return publish_map_version(**{**arguments, **overrides})
|
|
|
|
class Original:
|
|
calls = []
|
|
changed = False
|
|
|
|
def get(self, session_id):
|
|
assert session_id == source["session_id"]
|
|
return source
|
|
|
|
def bound(self, session_id, generation):
|
|
self.calls.append(generation)
|
|
if (
|
|
self.changed
|
|
or session_id != source["session_id"]
|
|
or generation != source["generation"]
|
|
):
|
|
raise ValueError("Original source changed")
|
|
return source
|
|
|
|
verify = bound
|
|
|
|
return SimpleNamespace(
|
|
root=tmp_path,
|
|
source=source,
|
|
points=points,
|
|
corrected=corrected,
|
|
publish=publish,
|
|
original=Original(),
|
|
)
|
|
|
|
|
|
def test_identity_idempotency_and_recomputed_distances(fixture):
|
|
f = fixture
|
|
version = f.publish()
|
|
assert f.publish().generation == version.generation
|
|
assert len(list((f.root / "versions").iterdir())) == 1
|
|
sources = VersionedPlanningSources(f.original, version, f.root / "scratch")
|
|
assert sources.get("source-a") is f.source
|
|
assert sources.bound("source-a", "a" * 64) is f.source
|
|
doc = sources.bound("source-a", version.generation)
|
|
assert doc["path_m"] == pytest.approx(2 * np.sqrt(101))
|
|
assert doc["path_m"] != f.source["path_m"]
|
|
assert doc["frame_id"] != f.source["frame_id"]
|
|
assert [p["message_index"] for p in doc["poses"]] == [1, 3, 5]
|
|
np.testing.assert_allclose([p["position"] for p in doc["poses"]], f.corrected)
|
|
assert doc["reference_version"]["authority"]["production_promotion"] is False
|
|
assert sources.version.arrays()["source_distance_m"][-1] == 20
|
|
# Caller mutation cannot affect the sealed artifact or its next projection.
|
|
doc["poses"][0]["position"][0] = 999
|
|
assert sources.bound("source-a", version.generation)["poses"][0]["position"][0] == 0
|
|
|
|
|
|
def test_same_session_generations_keep_old_and_new_drafts_distinct(fixture):
|
|
f = fixture
|
|
version = f.publish()
|
|
sources = VersionedPlanningSources(f.original, version, f.root / "scratch")
|
|
drafts = MissionDrafts(f.root / "drafts", sources)
|
|
|
|
def save(generation):
|
|
return drafts.save(
|
|
SimpleNamespace(
|
|
session_id="source-a",
|
|
generation=generation,
|
|
id=None,
|
|
revision=0,
|
|
name="Map test",
|
|
start_index=0,
|
|
end_index=2,
|
|
direction="forward",
|
|
)
|
|
)
|
|
|
|
old, new = save("a" * 64), save(version.generation)
|
|
assert old["route"]["length_m"] == 20
|
|
assert new["route"]["length_m"] == pytest.approx(2 * np.sqrt(101))
|
|
assert drafts.get(old["id"]) == old
|
|
assert drafts.check(new["id"], 1)["vehicle_control"] is False
|
|
assert drafts.check(new["id"], 1)["localization"] == "not_run"
|
|
|
|
|
|
def test_extraction_uses_corrected_frame_ownership_and_separate_display_profile(fixture):
|
|
f = fixture
|
|
version = f.publish()
|
|
sources = VersionedPlanningSources(f.original, version, f.root / "scratch")
|
|
points, provenance = sources.submap("source-a", version.generation, 0, 2)
|
|
np.testing.assert_allclose(points, f.corrected)
|
|
assert provenance["raw_points"] == 6 and provenance["retained_points"] == 3
|
|
assert [p["sequence"] for p in provenance["frames"]] == [10, 11, 12]
|
|
display, meta = sources.submap("source-a", version.generation, 0, 2, presentation=True)
|
|
np.testing.assert_allclose(display, f.points)
|
|
assert meta["extraction"]["radius_m"] == 80
|
|
assert meta["extraction"]["height_relative_m"] is None
|
|
whole, info = sources.reference_map("source-a", version.generation, 0, 2)
|
|
np.testing.assert_allclose(whole, f.corrected)
|
|
assert info["generation"] == version.generation
|
|
assert not list((f.root / "scratch").iterdir())
|
|
|
|
|
|
@pytest.mark.parametrize("target", ["manifest.json", "points.f32", "trajectory.npz", "review.json"])
|
|
def test_tampered_bundle_cannot_be_used(fixture, target):
|
|
f = fixture
|
|
version = f.publish()
|
|
path = version.directory / target
|
|
original = path.read_bytes()
|
|
path.write_bytes(b"X" + original[1:])
|
|
with pytest.raises(ValueError):
|
|
MapVersion(version.directory, version.generation).verify()
|
|
|
|
|
|
@pytest.mark.parametrize("mode", ["original", "derivative", "cancel", "exception"])
|
|
def test_preparation_is_private_and_postchecked(fixture, mode):
|
|
f = fixture
|
|
version = f.publish()
|
|
sources = VersionedPlanningSources(f.original, version, f.root / "scratch")
|
|
cancel = threading.Event()
|
|
with pytest.raises((ValueError, InterruptedError)), sources.prepared_submaps(
|
|
"source-a", version.generation, cancel_event=cancel
|
|
) as extract:
|
|
if mode == "original":
|
|
f.original.changed = True
|
|
elif mode == "derivative":
|
|
(version.directory / "points.f32").write_bytes(b"X" * 72)
|
|
# Still reads the private copy; final verification rejects publication.
|
|
np.testing.assert_allclose(extract(0, 2)[0], f.corrected)
|
|
elif mode == "cancel":
|
|
cancel.set()
|
|
extract(0, 2)
|
|
else:
|
|
raise ValueError("Synthetic preparation failure")
|
|
assert not list((f.root / "scratch").iterdir())
|
|
|
|
|
|
def test_wrong_parent_and_no_silent_fallback(fixture):
|
|
f = fixture
|
|
version = f.publish()
|
|
sources = VersionedPlanningSources(f.original, version, f.root / "scratch")
|
|
with pytest.raises(ValueError):
|
|
sources.bound("source-b", version.generation)
|
|
with pytest.raises(ValueError):
|
|
sources.bound("source-a", "c" * 64)
|
|
with pytest.raises(ValueError):
|
|
version.planning_source({**f.source, "generation": "d" * 64})
|
|
with pytest.raises(ValueError):
|
|
version.planning_source({**f.source, "source_digests": {"primary": "d" * 64}})
|
|
|
|
|
|
def test_publication_rejects_bad_hash_unsafe_names_nonfinite_points_and_links(fixture):
|
|
f = fixture
|
|
with pytest.raises(ValueError):
|
|
f.publish(expected_points_sha256="e" * 64)
|
|
with pytest.raises(ValueError):
|
|
f.publish(
|
|
evidence={"../escape.json": (f.root / "review.json", sha256(f.root / "review.json"))}
|
|
)
|
|
path = f.root / "link.f32"
|
|
path.symlink_to(f.root / "points.f32")
|
|
with pytest.raises(ValueError):
|
|
f.publish(points=path)
|
|
broken = f.points.copy()
|
|
broken[0, 0] = np.nan
|
|
broken.tofile(f.root / "points.f32")
|
|
with pytest.raises(ValueError):
|
|
f.publish()
|
|
assert not list((f.root / "versions").iterdir())
|
|
|
|
|
|
@pytest.mark.parametrize("mode", ["offset", "size", "clock", "source-distance", "quaternion"])
|
|
def test_publication_rejects_inconsistent_frame_pose_bindings(fixture, mode):
|
|
f = fixture
|
|
path = f.root / "trajectory.npz"
|
|
with np.load(path) as data:
|
|
values = {key: data[key].copy() for key in data.files}
|
|
if mode == "offset":
|
|
values["frames"][1, 2] = 1
|
|
elif mode == "size":
|
|
values["frames"][-1, 3] = 1
|
|
elif mode == "clock":
|
|
values["receipt_time_s"][1] = 4
|
|
elif mode == "source-distance":
|
|
values["frame_source_distance_m"][1] = 1
|
|
else:
|
|
values["orientations_xyzw"][0] = 0
|
|
np.savez(path, **values)
|
|
with pytest.raises(ValueError):
|
|
f.publish()
|
|
assert not list((f.root / "versions").iterdir())
|
|
|
|
|
|
def test_different_review_creates_new_version_without_overwrite(fixture):
|
|
f = fixture
|
|
first = f.publish()
|
|
(f.root / "review.json").write_text(json.dumps({"kind": "another-reviewed-candidate"}))
|
|
second = f.publish()
|
|
assert first.generation != second.generation
|
|
first.verify()
|
|
second.verify()
|