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.
248 lines
8.9 KiB
Python
248 lines
8.9 KiB
Python
"""Small failure fixtures; no device, network, live singleton or native search."""
|
|
|
|
from concurrent.futures import Future
|
|
from types import SimpleNamespace
|
|
|
|
import numpy as np
|
|
import pytest
|
|
from test_planning_live import event, fixture_service, until
|
|
from test_stream_summary import _pose_payload
|
|
|
|
from k1link.compute.live_perception import LivePerceptionIngress
|
|
from k1link.device_plugins.xgrids_k1.planning_live import K1PlanningLiveSource
|
|
from k1link.missions.stationary_live import run_stationary_live
|
|
|
|
|
|
@pytest.mark.parametrize("failure", ["persist", "thread-constructor", "thread-start", "open"])
|
|
def test_failed_start_releases_resources_and_allows_retry(tmp_path, monkeypatch, failure):
|
|
import k1link.missions.live_tests as module
|
|
|
|
service, source, compute_lock, _ = fixture_service(tmp_path, monkeypatch)
|
|
|
|
def fail(*args, **kwargs):
|
|
raise OSError("injected startup failure")
|
|
|
|
with monkeypatch.context() as patch:
|
|
if failure == "persist":
|
|
patch.setattr(service, "persist", fail)
|
|
elif failure == "thread-constructor":
|
|
patch.setattr(module.threading, "Thread", fail)
|
|
elif failure == "thread-start":
|
|
patch.setattr(module.threading.Thread, "start", fail)
|
|
else:
|
|
patch.setattr(source, "open", fail)
|
|
try:
|
|
with pytest.raises(OSError, match="injected startup failure"):
|
|
service.start("draft", 1)
|
|
assert not compute_lock.locked()
|
|
assert source.owner is None
|
|
assert service.thread is None
|
|
if failure != "open":
|
|
assert service.get()["state"] == "error"
|
|
assert service.get()["tracking_state"] == "lost"
|
|
assert service.source is None
|
|
finally:
|
|
# Also isolate the pre-fix red run, whose leaked resources are fixtures.
|
|
if compute_lock.locked():
|
|
compute_lock.release()
|
|
source.owner = None
|
|
|
|
run = service.start("draft", 1)
|
|
try:
|
|
until(lambda: service.get()["state"] == "waiting")
|
|
service.stop(run["id"])
|
|
finally:
|
|
service.close()
|
|
assert not compute_lock.locked() and source.owner is None
|
|
|
|
|
|
@pytest.mark.parametrize("failure", ["executor", "report", "close"])
|
|
def test_worker_cleanup_keeps_failure_terminal_and_unlocks(tmp_path, monkeypatch, failure):
|
|
import k1link.missions.live_tests as module
|
|
|
|
service, source, compute_lock, _ = fixture_service(tmp_path, monkeypatch)
|
|
original_persist = service.persist
|
|
original_close = source.close
|
|
|
|
def fail(*args, **kwargs):
|
|
raise OSError("injected worker failure")
|
|
|
|
if failure == "executor":
|
|
monkeypatch.setattr(module, "ThreadPoolExecutor", fail)
|
|
elif failure == "report":
|
|
|
|
def persist():
|
|
if service.run["state"] != "preparing":
|
|
fail()
|
|
original_persist()
|
|
|
|
monkeypatch.setattr(service, "persist", persist)
|
|
else:
|
|
|
|
def close(owner):
|
|
original_close(owner)
|
|
fail()
|
|
|
|
monkeypatch.setattr(source, "close", close)
|
|
|
|
run = service.start("draft", 1)
|
|
try:
|
|
if failure == "close":
|
|
until(lambda: service.get()["state"] == "waiting")
|
|
service.stop(run["id"])
|
|
until(lambda: not service.thread.is_alive())
|
|
assert not compute_lock.locked()
|
|
assert source.owner is None
|
|
assert service.get()["state"] == "error"
|
|
assert service.accepted_sample is None and service.last_result_ns == 0
|
|
finally:
|
|
service.close()
|
|
if compute_lock.locked():
|
|
compute_lock.release()
|
|
source.owner = None
|
|
|
|
|
|
def test_partial_report_write_is_recovered_as_terminal_before_retry(tmp_path, monkeypatch):
|
|
import k1link.missions.live_tests as module
|
|
|
|
service, source, compute_lock, _ = fixture_service(tmp_path, monkeypatch)
|
|
replace_file = module.os.replace
|
|
failures = []
|
|
|
|
def fail_first_active_replace(src, dst):
|
|
if dst == service.root / "active.json" and not failures:
|
|
failures.append(dst)
|
|
raise OSError("injected partial report commit")
|
|
replace_file(src, dst)
|
|
|
|
monkeypatch.setattr(module.os, "replace", fail_first_active_replace)
|
|
with pytest.raises(OSError, match="partial report commit"):
|
|
service.start("draft", 1)
|
|
assert not compute_lock.locked() and source.owner is None
|
|
assert service.thread is None
|
|
restored = module.PlanningLiveTests(service.drafts, service.sources, compute_lock)
|
|
assert restored.get()["state"] == "error"
|
|
assert restored.get()["tracking_state"] == "lost"
|
|
assert restored.get()["scene_available"] is False
|
|
assert (
|
|
"partial report commit"
|
|
in (service.directory(service.run["id"]) / "failure.txt").read_text()
|
|
)
|
|
run = service.start("draft", 1)
|
|
try:
|
|
until(lambda: service.get()["state"] == "waiting")
|
|
service.stop(run["id"])
|
|
finally:
|
|
service.close()
|
|
assert not compute_lock.locked() and source.owner is None
|
|
|
|
|
|
@pytest.mark.parametrize("modality", ["camera-init", "camera-frame"])
|
|
def test_auxiliary_receipt_is_not_an_empty_queue(modality):
|
|
ingress = LivePerceptionIngress()
|
|
source = K1PlanningLiveSource(ingress)
|
|
source.open("fixture")
|
|
ingress.begin_session("B")
|
|
try:
|
|
assert source.take("fixture").kind == "session-start"
|
|
ingress.publish(
|
|
modality=modality,
|
|
source_id="camera",
|
|
source_sequence=1,
|
|
captured_at_epoch_ns=10,
|
|
received_monotonic_ns=20,
|
|
payload=b"fixture",
|
|
)
|
|
ingress.publish(
|
|
modality="pose",
|
|
source_id="x/lio_pose",
|
|
source_sequence=2,
|
|
captured_at_epoch_ns=11,
|
|
received_monotonic_ns=21,
|
|
payload=_pose_payload((0, 0, 0)),
|
|
)
|
|
auxiliary = source.take("fixture")
|
|
assert auxiliary is not None and auxiliary.kind == "ignored"
|
|
assert auxiliary.session_id == "B" and auxiliary.generation == 1
|
|
assert auxiliary.monotonic_ns == 20 and auxiliary.epoch_ns == 10
|
|
assert auxiliary.points is None and auxiliary.position is None
|
|
assert ingress.snapshot()["queues"]["pose"]["depth"] == 1
|
|
pose = source.take("fixture")
|
|
assert pose.kind == "pose" and pose.sequence > auxiliary.sequence
|
|
finally:
|
|
source.close("fixture")
|
|
ingress.close()
|
|
|
|
|
|
def test_delayed_auxiliary_receipt_does_not_freeze_before_queued_prefix(tmp_path, monkeypatch):
|
|
service, _, _, _ = fixture_service(tmp_path, monkeypatch)
|
|
service.reference_path = np.array([[0.0, 0, 0], [20.0, 0, 0]])
|
|
points = np.random.default_rng(501).uniform([-2, -2, -1], [2, 2, 2], (700, 3))
|
|
service.reference = points
|
|
service.run = dict(
|
|
baseline_generation=0,
|
|
baseline_session_id="old",
|
|
draft=dict(zone=dict(session_id="A")),
|
|
maximum_distance_m=20,
|
|
)
|
|
updates = []
|
|
service.update = lambda **values: updates.append(values)
|
|
service.directory = lambda _: tmp_path
|
|
clock_value = [100.0]
|
|
pending = []
|
|
for t in range(100, 111):
|
|
pending.extend([event("pose", t=t), event("points", t=t + 0.001, points=points)])
|
|
if t == 108:
|
|
# Use the production adapter, not an invented None/ignored value.
|
|
ingress = LivePerceptionIngress()
|
|
adapter = K1PlanningLiveSource(ingress)
|
|
adapter.open("fixture")
|
|
ingress.begin_session("B")
|
|
adapter.take("fixture")
|
|
ingress.publish(
|
|
modality="camera-frame",
|
|
source_id="camera",
|
|
source_sequence=1,
|
|
captured_at_epoch_ns=1,
|
|
received_monotonic_ns=108_100_000_000,
|
|
payload=b"fixture",
|
|
)
|
|
pending.append(adapter.take("fixture"))
|
|
adapter.close("fixture")
|
|
ingress.close()
|
|
# Preserve the same session/generation/order across modality boundaries.
|
|
from dataclasses import replace
|
|
|
|
pending = [
|
|
(108.1, None)
|
|
if item is None
|
|
else (item.monotonic_ns / 1e9, replace(item, generation=1, sequence=i + 1))
|
|
for i, item in enumerate(pending)
|
|
]
|
|
|
|
class Source:
|
|
def snapshot(self):
|
|
return dict(active=True, session_id="B", session_generation=1)
|
|
|
|
def take(self, owner):
|
|
stamp, item = pending.pop(0)
|
|
clock_value[0] = stamp + 2
|
|
return item
|
|
|
|
searches = []
|
|
|
|
class Executor:
|
|
def submit(self, *args, **kwargs):
|
|
searches.append((args, kwargs))
|
|
service.cancel.set()
|
|
future = Future()
|
|
future.set_result({})
|
|
return future
|
|
|
|
clock = SimpleNamespace(
|
|
monotonic=lambda: clock_value[0], monotonic_ns=lambda: int(clock_value[0] * 1e9)
|
|
)
|
|
run_stationary_live(service, Source(), "fixture", Executor(), clock, None, None)
|
|
assert len(searches) == 1 and not pending
|
|
assert any(update.get("planning_phase") == "searching" for update in updates)
|