"""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)