DCCONSTRUCTIONS
e515ab1b8c
Preserve the completed teach-and-repeat laboratory stage: reference preparation, cascaded acquisition, local tracking and recovery, recording lifecycle, replay qualification, and persistent Rerun scene controls. Document the open grid-picking regression and Rerun upgrade contract. No autonomous driving or loop-closure optimization is claimed.
140 lines
5.5 KiB
Python
140 lines
5.5 KiB
Python
"""Exact local lookups and one verified preparation snapshot, not a load test."""
|
|
|
|
import json
|
|
import threading
|
|
from types import SimpleNamespace
|
|
|
|
import numpy as np
|
|
import pytest
|
|
from test_session_recording import _command
|
|
|
|
from k1link.missions.reference_map import build_reference_map
|
|
from k1link.missions.reference_window import ReferenceWindowIndex, reference_window
|
|
from k1link.missions.sources import PlanningSources
|
|
|
|
|
|
@pytest.mark.parametrize("center", [[0, 0, 0], [30, 5, -10], [-25, -20, 5]])
|
|
def test_index_preserves_exact_crop_order_duplicates_and_boundary(center):
|
|
rng = np.random.default_rng(601)
|
|
center = np.array(center, dtype=float)
|
|
near = rng.normal(center, 4, (1500, 3))
|
|
boundary = center + np.array([[10, 0, 0], [-10, 0, 0], [0, 0, 10]])
|
|
reference = np.vstack([near, boundary, near[:30], near + 150])
|
|
sample = dict(path=np.array([center]), points=np.array([center]))
|
|
expected, before = reference_window(reference, sample, np.eye(4))
|
|
index = ReferenceWindowIndex(reference)
|
|
actual, after = reference_window(reference, sample, np.eye(4), index=index)
|
|
np.testing.assert_array_equal(actual, expected)
|
|
assert before["target_sha256"] == after["target_sha256"]
|
|
assert after["examined_points"] < after["map_points"]
|
|
with pytest.raises(ValueError, match="another map"):
|
|
reference_window(reference.copy(), sample, np.eye(4), index=index)
|
|
|
|
|
|
def test_empty_index_and_broad_crop_are_exact():
|
|
index = ReferenceWindowIndex(np.empty((0, 3)))
|
|
assert index.crop(np.zeros(3), 10)[1] == 0
|
|
points = np.random.default_rng(4).normal(size=(400, 3))
|
|
index = ReferenceWindowIndex(points)
|
|
assert index.crop(np.zeros(3), 1000)[0] is points
|
|
with pytest.raises(ValueError):
|
|
index.crop(np.zeros(3), float("nan"))
|
|
|
|
|
|
def source_fixture(tmp_path, extractor):
|
|
command = _command(tmp_path / "source")
|
|
detail = SimpleNamespace(
|
|
plugin_id=command.plugin_id,
|
|
summary=SimpleNamespace(replayable=True, lab=None),
|
|
as_dict=lambda: {"display_name": "fixture"},
|
|
)
|
|
store = SimpleNamespace(
|
|
data_dir=tmp_path / "data", get_session=lambda _: detail, prepare_replay=lambda _: command
|
|
)
|
|
|
|
def export(_source, dest):
|
|
dest.write_text(json.dumps(dict(poses=[dict(distance_m=i) for i in range(101)])))
|
|
|
|
sources = PlanningSources(store, {command.plugin_id: export}, {command.plugin_id: extractor})
|
|
doc = sources.get(command.session_id)
|
|
return sources, command, doc["generation"]
|
|
|
|
|
|
@pytest.mark.parametrize("mode", ["success", "corrupt-source", "extractor-error", "cancel"])
|
|
def test_preparation_stages_once_checks_source_at_exit_and_cleans_up(tmp_path, monkeypatch, mode):
|
|
import k1link.missions.sources as module
|
|
|
|
paths = []
|
|
cancel = threading.Event()
|
|
|
|
def extract(path, doc, first, last):
|
|
paths.append(path)
|
|
if mode == "corrupt-source" and len(paths) == 2:
|
|
target = command.primary_artifact.path
|
|
target.write_bytes(b"X" * target.stat().st_size)
|
|
if mode == "extractor-error":
|
|
raise ValueError("fixture extraction failed")
|
|
if mode == "cancel":
|
|
cancel.set()
|
|
return np.array([[first, 0, 0], [last, 0, 0]], dtype=float), {"interval": [first, last]}
|
|
|
|
sources, command, generation = source_fixture(tmp_path, extract)
|
|
stages, validations = [], []
|
|
stage = module._stage_replay_prefix
|
|
verify = sources.verify
|
|
|
|
def counted_stage(*args, **kwargs):
|
|
result = stage(*args, **kwargs)
|
|
stages.append(result[0])
|
|
return result
|
|
|
|
def counted_verify(*args):
|
|
validations.append(args)
|
|
return verify(*args)
|
|
|
|
monkeypatch.setattr(module, "_stage_replay_prefix", counted_stage)
|
|
monkeypatch.setattr(sources, "verify", counted_verify)
|
|
if mode == "success":
|
|
points, meta = build_reference_map(sources, command.session_id, generation, 0, 100)
|
|
np.testing.assert_array_equal(points[:, 0], [0, 40, 80, 100])
|
|
assert len(meta["tiles"]) == 3 and len(validations) == 2
|
|
else:
|
|
with pytest.raises((ValueError, InterruptedError)):
|
|
build_reference_map(
|
|
sources, command.session_id, generation, 0, 100, cancel_event=cancel
|
|
)
|
|
assert len(stages) == 1 and not stages[0].exists()
|
|
assert len(set(paths)) == 1
|
|
assert not list(sources.root.glob(".source.*"))
|
|
|
|
|
|
def test_single_submap_retains_verified_api(tmp_path):
|
|
def extract(_path, _doc, first, last):
|
|
return np.ones((300, 3)), {"interval": [first, last]}
|
|
|
|
sources, command, generation = source_fixture(tmp_path, extract)
|
|
points, meta = sources.submap(command.session_id, generation, 1, 5)
|
|
assert len(points) == 300 and meta["interval"] == [1, 5]
|
|
assert meta["generation"] == generation
|
|
assert not list(sources.root.glob(".source.*"))
|
|
|
|
|
|
def test_cancellation_during_copy_remains_cancellation_not_a_product_error(tmp_path, monkeypatch):
|
|
import k1link.missions.sources as module
|
|
|
|
def unused(*args):
|
|
raise AssertionError("Cancelled snapshot must not reach the extractor.")
|
|
|
|
sources, command, generation = source_fixture(tmp_path, unused)
|
|
cancel = threading.Event()
|
|
stage = module._stage_replay_prefix
|
|
|
|
def cancel_during_stage(*args, **kwargs):
|
|
cancel.set()
|
|
return stage(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(module, "_stage_replay_prefix", cancel_during_stage)
|
|
with pytest.raises(InterruptedError, match="cancelled"):
|
|
build_reference_map(sources, command.session_id, generation, 0, 100, cancel_event=cancel)
|
|
assert not list(sources.root.glob(".source.*"))
|