from __future__ import annotations import asyncio import threading from pathlib import Path from typing import Any import pytest from pydantic import SecretStr, ValidationError import k1link.device_plugins.xgrids_k1.facade as facade_module from k1link.device_plugins.xgrids_k1.facade import ( DEFAULT_LIVE_STREAMS, XGRIDS_K1_COMPATIBILITY_PROFILE_ID, AbortAcquisitionRequest, CameraPreviewSelectRequest, CompatibilityAttestationRequest, ConnectRequest, OpenApplicationControlSessionRequest, OperatorPresenceRequest, PrepareAcquisitionRequest, ShadowApplicationControlArmRequest, StartAcquisitionRequest, StopAcquisitionRequest, XgridsK1CompatibilityService, ) from k1link.device_plugins.xgrids_k1.protocol.application_bootstrap import ( ApplicationControlAuthority, ) ATTESTATION = CompatibilityAttestationRequest( firmware_version="3.0.2", topology="direct-lan", operator_confirmed=True, ) PRIMARY_TEST_CREDENTIAL = "x" * 24 SECONDARY_TEST_CREDENTIAL = "y" * 24 PROJECT_NAME = "K1 lifecycle test" PRIVATE_APPLICATION_AUTHORITY = "11111111-2222-3333-4444-555555555555" class FakeApplicationAuthorityLoader: def __init__(self) -> None: self.calls = 0 def load(self) -> ApplicationControlAuthority: self.calls += 1 return ApplicationControlAuthority(openapi_key=PRIVATE_APPLICATION_AUTHORITY) class FakeVisualizationRuntime: def __init__(self) -> None: self.phase = "idle" self.source_mode = "idle" self.source_ready = False self.pcl_frames = 0 self.start_calls: list[tuple[str, Path, float, str]] = [] self.stop_calls = 0 self.stop_error: Exception | None = None def snapshot(self) -> dict[str, Any]: return { "phase": self.phase, "message": "test runtime", "source_mode": self.source_mode, "source_ready": self.source_ready, "foxglove_ws_url": None, "foxglove_viewer_url": None, "rerun_grpc_url": None, "viewer_settings": {}, "metrics": { "messages_received": self.pcl_frames, "payload_bytes": 0, "pcl_frames": self.pcl_frames, "pose_frames": 0, "points_published": 0, "last_point_count": 0, "decode_errors": 0, "preview_dropped": 0, "pcl_fps": 0.0, "pose_fps": 0.0, "mqtt_to_publish_ms": None, "mqtt_to_publish_p50_ms": None, "mqtt_to_publish_p95_ms": None, "decode_publish_ms": None, "trajectory_poses": 0, }, } def start_live( self, host: str, out_dir: Path, *, duration_seconds: float, project_name: str, ) -> None: self.start_calls.append((host, out_dir, duration_seconds, project_name)) self.phase = "starting_live" self.source_mode = "live" def mark_ready(self) -> None: self.phase = "live" self.source_ready = True def stop(self) -> None: self.stop_calls += 1 if self.stop_error is not None: raise self.stop_error self.phase = "idle" self.source_mode = "idle" self.source_ready = False def close(self) -> None: self.stop() class FakeInteractiveControlSession: def __init__(self) -> None: self.state = "workspace-ready" self.start_projects: list[str] = [] self.stop_calls = 0 self.confirm_calls = 0 def snapshot(self) -> dict[str, object]: return { "state": self.state, "can_confirm_standby": self.state == "awaiting-standby-confirmation", } def open_project_prompt(self) -> dict[str, object]: assert self.state == "workspace-ready" self.state = "project-ready" return self.snapshot() def request_start(self, *, project_name: str, confirmation: object) -> dict[str, object]: assert confirmation is not None assert self.state == "project-ready" self.start_projects.append(project_name) self.state = "scanning" return self.snapshot() def request_stop(self, *, confirmation: object) -> dict[str, object]: assert confirmation is not None assert self.state == "scanning" self.stop_calls += 1 self.state = "awaiting-standby-confirmation" return self.snapshot() def confirm_standby(self) -> dict[str, object]: assert self.state == "awaiting-standby-confirmation" self.confirm_calls += 1 self.state = "completed" return self.snapshot() def close_prestart(self) -> dict[str, object]: self.state = "closed" return self.snapshot() def close(self) -> None: self.state = "closed" PHYSICAL_ACCEPTANCE = OperatorPresenceRequest( operator_present=True, owner_controlled_device=True, lixelgo_closed=True, battery_storage_confirmed=True, expected_physical_state_confirmed=True, ) def test_control_session_transport_metadata_does_not_leak_into_confirmation() -> None: request = OpenApplicationControlSessionRequest( operator_present=True, owner_controlled_device=True, lixelgo_closed=True, battery_storage_confirmed=True, expected_physical_state_confirmed=True, timezone_name="Europe/Moscow", ) assert request.confirmation() == PHYSICAL_ACCEPTANCE.confirmation() def service_with_fake_runtime( tmp_path: Path, ) -> tuple[XgridsK1CompatibilityService, FakeVisualizationRuntime]: service = XgridsK1CompatibilityService(tmp_path) runtime = FakeVisualizationRuntime() service.runtime = runtime # type: ignore[assignment] return service, runtime def test_prepare_creates_provisional_device_session_and_profiled_acquisition( tmp_path: Path, ) -> None: service, runtime = service_with_fake_runtime(tmp_path) state = service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", duration_seconds=60, compatibility_attestation=ATTESTATION, ) ) assert runtime.start_calls == [] assert state["device_ref"]["identity_stability"] == "provisional" assert state["device_ref"]["device_id"] != state["device_session"]["device_session_id"] assert state["acquisition"]["state"] == "prepared" assert state["acquisition"]["project_name"] == PROJECT_NAME assert state["acquisition"]["compatibility_profile_id"] == (XGRIDS_K1_COMPATIBILITY_PROFILE_ID) assert state["compatibility"]["vendor_writes_enabled"] is False def test_project_name_is_normalized_and_control_characters_are_rejected() -> None: request = PrepareAcquisitionRequest( project_name=" Ｋ１ Lab ", host="192.168.1.20", compatibility_attestation=ATTESTATION, ) assert request.project_name == "K1 Lab" for invalid in (" ", "line\nbreak", "\ud800", "x" * 97): with pytest.raises(ValidationError): PrepareAcquisitionRequest( project_name=invalid, host="192.168.1.20", compatibility_attestation=ATTESTATION, ) def test_facade_arms_bounded_shadow_lease_without_installing_publish_transport( tmp_path: Path, ) -> None: loader = FakeApplicationAuthorityLoader() service = XgridsK1CompatibilityService( tmp_path, application_authority_loader=loader, ) runtime = FakeVisualizationRuntime() service.runtime = runtime # type: ignore[assignment] service._selected_device_id = "test-ble-transport" # noqa: SLF001 service._k1_ip = "192.168.1.20" # noqa: SLF001 service._compatibility_attestation = { # noqa: SLF001 "firmware_version": "3.0.2", "topology": "direct-lan", "basis": "operator-attested", "observed_at": "2026-07-18T00:00:00Z", } state = service.arm_application_control_shadow( ShadowApplicationControlArmRequest( operator_confirmed=True, lease_seconds=60.0, timezone_name="Europe/Moscow", ) ) execution = state["application_control_execution"] assert loader.calls == 1 assert execution["state"] == "armed-shadow-only" assert execution["lease"]["authority_cached"] is True assert execution["can_emit_requests"] is False assert execution["live_transport_installed"] is False assert execution["publisher"]["vendor_writes_enabled"] is False assert execution["publisher"]["transport_calls"] == 0 assert PRIVATE_APPLICATION_AUTHORITY not in str(state) disarmed = service.disarm_application_control_shadow() assert disarmed["application_control_execution"]["state"] == "disarmed" assert disarmed["application_control_execution"]["lease"] is None def test_shadow_arm_requires_idle_connected_attested_device_before_keychain_read( tmp_path: Path, ) -> None: loader = FakeApplicationAuthorityLoader() service = XgridsK1CompatibilityService( tmp_path, application_authority_loader=loader, ) service.runtime = FakeVisualizationRuntime() # type: ignore[assignment] with pytest.raises(RuntimeError, match="подключите K1"): service.arm_application_control_shadow( ShadowApplicationControlArmRequest( operator_confirmed=True, timezone_name="Europe/Moscow", ) ) assert loader.calls == 0 def test_shadow_arm_contract_requires_explicit_operator_confirmation() -> None: with pytest.raises(ValidationError): ShadowApplicationControlArmRequest.model_validate({"timezone_name": "Europe/Moscow"}) def test_prepare_acquisition_revokes_existing_shadow_authority_lease(tmp_path: Path) -> None: loader = FakeApplicationAuthorityLoader() service = XgridsK1CompatibilityService( tmp_path, application_authority_loader=loader, ) service.runtime = FakeVisualizationRuntime() # type: ignore[assignment] service._selected_device_id = "test-ble-transport" # noqa: SLF001 service._k1_ip = "192.168.1.20" # noqa: SLF001 service._compatibility_attestation = {"profile": "exact"} # noqa: SLF001 service.arm_application_control_shadow( ShadowApplicationControlArmRequest( operator_confirmed=True, timezone_name="Europe/Moscow", ) ) prepared = service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", duration_seconds=60, compatibility_attestation=ATTESTATION, ) ) assert prepared["application_control_execution"]["state"] == "disarmed" assert prepared["application_control_execution"]["lease"] is None def test_operator_manual_start_is_confirmed_only_by_real_point_data(tmp_path: Path) -> None: service, runtime = service_with_fake_runtime(tmp_path) prepared = service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", duration_seconds=60, compatibility_attestation=ATTESTATION, ) ) acquisition_id = prepared["acquisition"]["acquisition_id"] starting = service.start_acquisition(StartAcquisitionRequest(acquisition_id=acquisition_id)) assert starting["acquisition"]["state"] == "starting" assert starting["last_operation"]["status"] == "running" runtime.mark_ready() awaiting = service.state() assert awaiting["acquisition"]["state"] == "awaiting_external_start" assert awaiting["last_operation"]["status"] == "operator_action_required" assert len(runtime.start_calls) == 1 assert runtime.start_calls[0][3] == PROJECT_NAME runtime.pcl_frames = 1 acquiring = service.state() assert acquiring["acquisition"]["state"] == "acquiring" assert acquiring["last_operation"]["status"] == "succeeded" assert acquiring["last_operation"]["result"]["confirmation"] == "point-frame" def test_plugin_commanded_acquisition_keeps_start_and_stop_as_explicit_actions( tmp_path: Path, ) -> None: service, runtime = service_with_fake_runtime(tmp_path) control = FakeInteractiveControlSession() service._application_control_session = control # type: ignore[assignment] # noqa: SLF001 service._k1_ip = "192.168.1.20" # noqa: SLF001 service._compatibility_attestation = {"profile": "exact"} # noqa: SLF001 prepared = service.prepare_acquisition( PrepareAcquisitionRequest( project_name="TEST001", host="192.168.1.20", compatibility_attestation=ATTESTATION, ) ) acquisition_id = prepared["acquisition"]["acquisition_id"] assert prepared["acquisition"]["control_mode"] == "plugin-commanded" assert control.state == "project-ready" assert runtime.start_calls == [] with pytest.raises(ValueError, match="подтверждения присутствия"): service.start_acquisition(StartAcquisitionRequest(acquisition_id=acquisition_id)) assert control.start_projects == [] service.start_acquisition( StartAcquisitionRequest( acquisition_id=acquisition_id, physical_acceptance=PHYSICAL_ACCEPTANCE, ) ) assert control.start_projects == ["TEST001"] runtime.mark_ready() runtime.pcl_frames = 1 acquiring = service.state() assert acquiring["acquisition"]["state"] == "acquiring" stopping = service.stop_acquisition( StopAcquisitionRequest( acquisition_id=acquisition_id, mode="graceful", physical_acceptance=PHYSICAL_ACCEPTANCE, ) ) stop_operation = stopping["last_operation"] assert control.stop_calls == 1 assert stopping["acquisition"]["state"] == "awaiting_external_stop" completed = service.stop_acquisition( StopAcquisitionRequest( acquisition_id=acquisition_id, mode="graceful", operator_confirmed=True, operation_id=stop_operation["operation_id"], ) ) assert control.stop_calls == 1 assert control.confirm_calls == 1 assert completed["acquisition"]["state"] == "completed" assert runtime.stop_calls == 1 def test_second_start_conflict_does_not_tear_down_start_owner(tmp_path: Path) -> None: service, runtime = service_with_fake_runtime(tmp_path) prepared = service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", compatibility_attestation=ATTESTATION, ) ) acquisition_id = prepared["acquisition"]["acquisition_id"] service.start_acquisition(StartAcquisitionRequest(acquisition_id=acquisition_id)) with pytest.raises(RuntimeError, match="состояния starting"): service.start_acquisition(StartAcquisitionRequest(acquisition_id=acquisition_id)) state = service.state() start_operations = [ item for item in state["operations"] if item["action"] == "acquisition.start" ] assert state["acquisition"]["state"] == "starting" assert runtime.stop_calls == 0 assert [item["status"] for item in start_operations] == ["running", "failed"] assert start_operations[-1]["error"]["category"] == "conflict" def test_state_waits_for_atomic_start_handoff( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: service, runtime = service_with_fake_runtime(tmp_path) prepared = service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", compatibility_attestation=ATTESTATION, ) ) acquisition_id = prepared["acquisition"]["acquisition_id"] entered_start = threading.Event() release_start = threading.Event() state_started = threading.Event() state_finished = threading.Event() worker_errors: list[BaseException] = [] snapshots: list[dict[str, Any]] = [] original_start_live = runtime.start_live def blocked_start_live(*args: object, **kwargs: object) -> None: entered_start.set() if not release_start.wait(timeout=2): raise TimeoutError("test did not release start handoff") original_start_live(*args, **kwargs) # type: ignore[arg-type] def start_worker() -> None: try: service.start_acquisition(StartAcquisitionRequest(acquisition_id=acquisition_id)) except BaseException as exc: # pragma: no cover - asserted below worker_errors.append(exc) def state_worker() -> None: state_started.set() try: snapshots.append(service.state()) except BaseException as exc: # pragma: no cover - asserted below worker_errors.append(exc) finally: state_finished.set() monkeypatch.setattr(runtime, "start_live", blocked_start_live) start_thread = threading.Thread(target=start_worker) state_thread = threading.Thread(target=state_worker) start_thread.start() assert entered_start.wait(timeout=2) state_thread.start() assert state_started.wait(timeout=2) assert not state_finished.wait(timeout=0.05) release_start.set() start_thread.join(timeout=2) state_thread.join(timeout=2) assert not start_thread.is_alive() assert not state_thread.is_alive() assert worker_errors == [] assert snapshots[0]["acquisition"]["state"] == "starting" assert runtime.source_mode == "live" assert service._acquisition_session_lease is not None # noqa: SLF001 def test_abort_waits_for_start_handoff_then_stops_owned_producers( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: service, runtime = service_with_fake_runtime(tmp_path) prepared = service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", compatibility_attestation=ATTESTATION, ) ) acquisition_id = prepared["acquisition"]["acquisition_id"] entered_start = threading.Event() release_start = threading.Event() abort_finished = threading.Event() worker_errors: list[BaseException] = [] original_start_live = runtime.start_live def blocked_start_live(*args: object, **kwargs: object) -> None: entered_start.set() if not release_start.wait(timeout=2): raise TimeoutError("test did not release start handoff") original_start_live(*args, **kwargs) # type: ignore[arg-type] def start_worker() -> None: try: service.start_acquisition(StartAcquisitionRequest(acquisition_id=acquisition_id)) except BaseException as exc: # pragma: no cover - asserted below worker_errors.append(exc) def abort_worker() -> None: try: service.abort_acquisition(AbortAcquisitionRequest(acquisition_id=acquisition_id)) except BaseException as exc: # pragma: no cover - asserted below worker_errors.append(exc) finally: abort_finished.set() monkeypatch.setattr(runtime, "start_live", blocked_start_live) start_thread = threading.Thread(target=start_worker) abort_thread = threading.Thread(target=abort_worker) start_thread.start() assert entered_start.wait(timeout=2) abort_thread.start() assert not abort_finished.wait(timeout=0.05) release_start.set() start_thread.join(timeout=2) abort_thread.join(timeout=2) assert worker_errors == [] state = service.state() assert state["acquisition"]["state"] == "aborted" assert runtime.source_mode == "idle" assert service._acquisition_session_lease is None # noqa: SLF001 def test_camera_selection_finishes_before_serialized_acquisition_stop( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: service, runtime = service_with_fake_runtime(tmp_path) prepared = service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", compatibility_attestation=ATTESTATION, ) ) acquisition_id = prepared["acquisition"]["acquisition_id"] device_session_id = prepared["device_session"]["device_session_id"] service.start_acquisition(StartAcquisitionRequest(acquisition_id=acquisition_id)) service._k1_ip = "192.168.1.20" # noqa: SLF001 entered_camera_arm = threading.Event() release_camera_arm = threading.Event() stop_finished = threading.Event() worker_errors: list[BaseException] = [] events: list[str] = [] def blocked_camera_arm(_out_dir: Path, *, require_session: bool = False) -> None: assert require_session is True events.append("arm") entered_camera_arm.set() if not release_camera_arm.wait(timeout=2): raise TimeoutError("test did not release camera arm") def select_worker() -> None: try: service.select_camera_preview( CameraPreviewSelectRequest( source_id="sensor.camera.left", device_session_id=device_session_id, ) ) except BaseException as exc: # pragma: no cover - asserted below worker_errors.append(exc) def stop_worker() -> None: try: service.stop_acquisition( StopAcquisitionRequest( acquisition_id=acquisition_id, mode="capture-only", ) ) except BaseException as exc: # pragma: no cover - asserted below worker_errors.append(exc) finally: stop_finished.set() monkeypatch.setattr(service, "_arm_camera_recording", blocked_camera_arm) monkeypatch.setattr( service.camera_preview, "select", lambda _source_id, _target: events.append("select"), ) monkeypatch.setattr( service.camera_preview, "stop_recording", lambda **_kwargs: events.append("camera-stop"), ) monkeypatch.setattr(runtime, "stop", lambda: events.append("runtime-stop")) select_thread = threading.Thread(target=select_worker) stop_thread = threading.Thread(target=stop_worker) select_thread.start() assert entered_camera_arm.wait(timeout=2) stop_thread.start() assert not stop_finished.wait(timeout=0.05) release_camera_arm.set() select_thread.join(timeout=2) stop_thread.join(timeout=2) assert worker_errors == [] assert events == ["arm", "select", "camera-stop", "runtime-stop"] assert service.state()["acquisition"]["state"] == "completed" assert service._acquisition_session_lease is None # noqa: SLF001 def test_camera_arm_failure_seals_stopped_session_before_releasing_lease( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: service, runtime = service_with_fake_runtime(tmp_path) prepared = service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", compatibility_attestation=ATTESTATION, ) ) acquisition_id = prepared["acquisition"]["acquisition_id"] events: list[str] = [] def start_live( _host: str, out_dir: Path, *, duration_seconds: float, project_name: str, ) -> None: assert duration_seconds > 0 assert project_name == PROJECT_NAME events.append("start") out_dir.mkdir(parents=True) runtime.phase = "starting_live" runtime.source_mode = "live" def stop_runtime() -> None: events.append("runtime") runtime.phase = "idle" runtime.source_mode = "idle" monkeypatch.setattr(runtime, "start_live", start_live) monkeypatch.setattr(runtime, "stop", stop_runtime) monkeypatch.setattr( service, "_arm_camera_recording", lambda _out_dir: (_ for _ in ()).throw(RuntimeError("synthetic camera arm failure")), ) monkeypatch.setattr( service.camera_preview, "stop_recording", lambda **_kwargs: events.append("camera"), ) monkeypatch.setattr( facade_module, "seal_capture_clock", lambda _capture_root: events.append("seal"), ) with pytest.raises(RuntimeError, match="synthetic camera arm failure"): service.start_acquisition(StartAcquisitionRequest(acquisition_id=acquisition_id)) state = service.state() assert state["acquisition"]["state"] == "failed" assert state["last_operation"]["status"] == "failed" assert events == ["start", "camera", "runtime", "seal"] assert service._acquisition_session_lease is None # noqa: SLF001 def test_start_cleanup_failure_retains_lease_and_marks_side_effect_unknown( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: service, runtime = service_with_fake_runtime(tmp_path) prepared = service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", compatibility_attestation=ATTESTATION, ) ) acquisition_id = prepared["acquisition"]["acquisition_id"] def fail_camera_arm(out_dir: Path) -> None: out_dir.mkdir() raise RuntimeError("synthetic camera arm failure") monkeypatch.setattr(service, "_arm_camera_recording", fail_camera_arm) monkeypatch.setattr(facade_module, "seal_capture_clock", lambda _capture_root: None) runtime.stop_error = RuntimeError("synthetic cleanup timeout") with pytest.raises(RuntimeError, match="synthetic camera arm failure"): service.start_acquisition(StartAcquisitionRequest(acquisition_id=acquisition_id)) state = service.state() operation = next(item for item in state["operations"] if item["action"] == "acquisition.start") assert state["acquisition"]["state"] == "failed" assert operation["status"] == "failed" assert operation["error"]["side_effect_status"] == "unknown" assert service._acquisition_session_lease is not None # noqa: SLF001 runtime.stop_error = None service.stop() assert service._acquisition_session_lease is None # noqa: SLF001 def test_receiver_completion_without_point_data_fails_start_operation(tmp_path: Path) -> None: service, runtime = service_with_fake_runtime(tmp_path) prepared = service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", duration_seconds=60, compatibility_attestation=ATTESTATION, ) ) acquisition_id = prepared["acquisition"]["acquisition_id"] service.start_acquisition(StartAcquisitionRequest(acquisition_id=acquisition_id)) runtime.phase = "idle" runtime.source_mode = "idle" failed = service.state() assert failed["acquisition"]["state"] == "failed" assert failed["last_operation"]["status"] == "failed" assert failed["acquisition"]["result"]["device_state"] == "unknown" def test_capture_only_stop_never_claims_that_physical_k1_stopped(tmp_path: Path) -> None: service, runtime = service_with_fake_runtime(tmp_path) prepared = service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", duration_seconds=60, compatibility_attestation=ATTESTATION, ) ) acquisition_id = prepared["acquisition"]["acquisition_id"] service.start_acquisition(StartAcquisitionRequest(acquisition_id=acquisition_id)) stopped = service.stop_acquisition( StopAcquisitionRequest(acquisition_id=acquisition_id, mode="capture-only") ) assert runtime.stop_calls == 1 assert stopped["acquisition"]["state"] == "completed" assert stopped["acquisition"]["result"] == { "receiver_stopped": True, "device_stop": "unknown", } operations = {item["action"]: item for item in stopped["operations"]} assert operations["acquisition.start"]["status"] == "cancelled" assert operations["acquisition.stop"]["status"] == "succeeded" def test_stop_seals_session_clock_after_camera_and_runtime_before_lease_release( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: service, runtime = service_with_fake_runtime(tmp_path) out_dir = tmp_path / "evidence-session" out_dir.mkdir() service._acquisition_out_dir = out_dir # noqa: SLF001 events: list[object] = [] monkeypatch.setattr( service.camera_preview, "stop_recording", lambda **_kwargs: events.append("camera"), ) monkeypatch.setattr(runtime, "stop", lambda: events.append("runtime")) monkeypatch.setattr( facade_module, "seal_capture_clock", lambda capture_root: events.append(("seal", capture_root)), ) class Lease: def release(self) -> None: events.append("lease") service._acquisition_session_lease = Lease() # type: ignore[assignment] # noqa: SLF001 service._stop_acquisition_sources( # noqa: SLF001 camera_status="complete", camera_failure_code=None, ) assert events == [ "camera", "runtime", ("seal", out_dir / "captures" / "mqtt_live"), "lease", ] def test_stop_timeout_retains_lease_and_blocks_replacement_acquisition( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: service, runtime = service_with_fake_runtime(tmp_path) prepared = service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", compatibility_attestation=ATTESTATION, ) ) acquisition_id = prepared["acquisition"]["acquisition_id"] service.start_acquisition(StartAcquisitionRequest(acquisition_id=acquisition_id)) out_dir = service._acquisition_out_dir # noqa: SLF001 assert out_dir is not None out_dir.mkdir() seal_calls: list[Path] = [] monkeypatch.setattr( facade_module, "seal_capture_clock", lambda capture_root: seal_calls.append(capture_root), ) runtime.stop_error = RuntimeError("synthetic stop timeout") with pytest.raises(RuntimeError, match="synthetic stop timeout"): service.stop_acquisition( StopAcquisitionRequest(acquisition_id=acquisition_id, mode="capture-only") ) assert seal_calls == [] assert service._acquisition_session_lease is not None # noqa: SLF001 assert service.state()["acquisition"]["cleanup_pending"] is True with pytest.raises(RuntimeError, match="evidence-сессия"): service.prepare_acquisition( PrepareAcquisitionRequest( project_name="replacement", host="192.168.1.20", compatibility_attestation=ATTESTATION, ) ) runtime.stop_error = None retried = service.stop() assert retried["acquisition"]["state"] == "failed" assert len(seal_calls) == 1 assert service._acquisition_session_lease is None # noqa: SLF001 assert retried["acquisition"]["cleanup_pending"] is False replacement = service.prepare_acquisition( PrepareAcquisitionRequest( project_name="replacement", host="192.168.1.20", compatibility_attestation=ATTESTATION, ) ) assert replacement["acquisition"]["state"] == "prepared" def test_replay_rejects_retained_failed_acquisition_lease( tmp_path: Path, ) -> None: service, runtime = service_with_fake_runtime(tmp_path) prepared = service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", compatibility_attestation=ATTESTATION, ) ) acquisition_id = prepared["acquisition"]["acquisition_id"] service.start_acquisition(StartAcquisitionRequest(acquisition_id=acquisition_id)) runtime.stop_error = RuntimeError("synthetic stop timeout") with pytest.raises(RuntimeError, match="synthetic stop timeout"): service.stop_acquisition( StopAcquisitionRequest(acquisition_id=acquisition_id, mode="capture-only") ) with pytest.raises(RuntimeError, match="не запечатана"): service.start_replay("sessions/fixture.k1mqtt", speed=1.0, loop=False) runtime.stop_error = None service.stop() def test_new_explicit_stop_retries_retained_terminal_cleanup( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: service, runtime = service_with_fake_runtime(tmp_path) prepared = service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", compatibility_attestation=ATTESTATION, ) ) acquisition_id = prepared["acquisition"]["acquisition_id"] service.start_acquisition(StartAcquisitionRequest(acquisition_id=acquisition_id)) out_dir = service._acquisition_out_dir # noqa: SLF001 assert out_dir is not None out_dir.mkdir() seal_calls: list[Path] = [] monkeypatch.setattr( facade_module, "seal_capture_clock", lambda capture_root: seal_calls.append(capture_root), ) runtime.stop_error = RuntimeError("synthetic stop timeout") with pytest.raises(RuntimeError, match="synthetic stop timeout"): service.stop_acquisition( StopAcquisitionRequest(acquisition_id=acquisition_id, mode="capture-only") ) runtime.stop_error = None recovered = service.stop_acquisition( StopAcquisitionRequest( acquisition_id=acquisition_id, mode="capture-only", idempotency_key="retry-retained-cleanup", ) ) retry_operation = recovered["last_operation"] assert retry_operation["status"] == "succeeded" assert retry_operation["stage_code"] == "retained-cleanup-completed" assert retry_operation["result"]["device_stop"] == "unknown" assert len(seal_calls) == 1 assert service._acquisition_session_lease is None # noqa: SLF001 def test_natural_receiver_completion_fails_acquisition_when_session_clock_cannot_seal( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: service, runtime = service_with_fake_runtime(tmp_path) prepared = service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", compatibility_attestation=ATTESTATION, ) ) acquisition_id = prepared["acquisition"]["acquisition_id"] service.start_acquisition(StartAcquisitionRequest(acquisition_id=acquisition_id)) runtime.pcl_frames = 1 service.state() out_dir = service._acquisition_out_dir # noqa: SLF001 assert out_dir is not None out_dir.mkdir() monkeypatch.setattr( facade_module, "seal_capture_clock", lambda _capture_root: (_ for _ in ()).throw(RuntimeError("synthetic seal failure")), ) runtime.phase = "idle" runtime.source_mode = "idle" with pytest.raises(RuntimeError, match="synthetic seal failure"): service.state() assert service._acquisition is not None # noqa: SLF001 assert service._acquisition.state == "failed" # noqa: SLF001 def test_natural_receiver_completion_reserves_finalization_before_reentrant_callback( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: service, runtime = service_with_fake_runtime(tmp_path) prepared = service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", compatibility_attestation=ATTESTATION, ) ) acquisition_id = prepared["acquisition"]["acquisition_id"] service.start_acquisition(StartAcquisitionRequest(acquisition_id=acquisition_id)) runtime.pcl_frames = 1 service.state() out_dir = service._acquisition_out_dir # noqa: SLF001 assert out_dir is not None out_dir.mkdir() events: list[str] = [] def reentrant_camera_stop(**_kwargs: object) -> None: events.append("camera") nested = service.state() assert nested["acquisition"]["state"] == "finalizing" monkeypatch.setattr(service.camera_preview, "stop_recording", reentrant_camera_stop) monkeypatch.setattr( facade_module, "seal_capture_clock", lambda _capture_root: events.append("seal"), ) runtime.phase = "idle" runtime.source_mode = "idle" completed = service.state() assert completed["acquisition"]["state"] == "completed" assert events == ["camera", "seal"] def test_graceful_stop_waits_for_explicit_operator_confirmation(tmp_path: Path) -> None: service, runtime = service_with_fake_runtime(tmp_path) prepared = service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", duration_seconds=60, compatibility_attestation=ATTESTATION, ) ) acquisition_id = prepared["acquisition"]["acquisition_id"] service.start_acquisition(StartAcquisitionRequest(acquisition_id=acquisition_id)) runtime.mark_ready() service.state() runtime.pcl_frames = 1 acquiring = service.state() assert acquiring["acquisition"]["state"] == "acquiring" awaiting = service.stop_acquisition( StopAcquisitionRequest(acquisition_id=acquisition_id, mode="graceful") ) operation_id = awaiting["last_operation"]["operation_id"] assert runtime.stop_calls == 0 assert awaiting["acquisition"]["state"] == "awaiting_external_stop" assert awaiting["last_operation"]["status"] == "operator_action_required" retried = service.stop_acquisition( StopAcquisitionRequest( acquisition_id=acquisition_id, operation_id=operation_id, mode="graceful", ) ) assert retried["acquisition"]["state"] == "awaiting_external_stop" assert retried["last_operation"]["operation_id"] == operation_id assert retried["last_operation"]["status"] == "operator_action_required" with pytest.raises(ValueError, match="исходную stop-operation"): service.stop_acquisition( StopAcquisitionRequest( acquisition_id=acquisition_id, mode="graceful", operator_confirmed=True, ) ) completed = service.stop_acquisition( StopAcquisitionRequest( acquisition_id=acquisition_id, operation_id=operation_id, mode="graceful", operator_confirmed=True, ) ) assert runtime.stop_calls == 1 assert completed["acquisition"]["result"]["device_stop"] == "operator-confirmed" stop_operations = [ item for item in completed["operations"] if item["action"] == "acquisition.stop" ] assert len(stop_operations) == 1 assert stop_operations[0]["operation_id"] == operation_id assert stop_operations[0]["status"] == "succeeded" def test_graceful_stop_retry_by_idempotency_key_reuses_original_operation( tmp_path: Path, ) -> None: service, runtime = service_with_fake_runtime(tmp_path) prepared = service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", compatibility_attestation=ATTESTATION, ) ) acquisition_id = prepared["acquisition"]["acquisition_id"] service.start_acquisition(StartAcquisitionRequest(acquisition_id=acquisition_id)) runtime.mark_ready() service.state() runtime.pcl_frames = 1 service.state() first = service.stop_acquisition( StopAcquisitionRequest( acquisition_id=acquisition_id, idempotency_key="graceful-stop-once", mode="graceful", ) ) first_operation_id = first["last_operation"]["operation_id"] retried = service.stop_acquisition( StopAcquisitionRequest( acquisition_id=acquisition_id, idempotency_key="graceful-stop-once", mode="graceful", ) ) assert retried["last_operation"]["operation_id"] == first_operation_id assert retried["last_operation"]["status"] == "operator_action_required" assert ( len([item for item in retried["operations"] if item["action"] == "acquisition.stop"]) == 1 ) def test_unrelated_graceful_stop_is_rejected_while_confirmation_is_pending( tmp_path: Path, ) -> None: service, runtime = service_with_fake_runtime(tmp_path) prepared = service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", compatibility_attestation=ATTESTATION, ) ) acquisition_id = prepared["acquisition"]["acquisition_id"] service.start_acquisition(StartAcquisitionRequest(acquisition_id=acquisition_id)) runtime.mark_ready() service.state() runtime.pcl_frames = 1 service.state() first = service.stop_acquisition( StopAcquisitionRequest(acquisition_id=acquisition_id, mode="graceful") ) expected_operation_id = first["last_operation"]["operation_id"] with pytest.raises(ValueError, match="уже ожидает"): service.stop_acquisition( StopAcquisitionRequest( acquisition_id=acquisition_id, idempotency_key="unrelated-stop", mode="graceful", ) ) original = next( item for item in service.state()["operations"] if item["operation_id"] == expected_operation_id ) assert original["status"] == "operator_action_required" def test_graceful_stop_is_rejected_until_point_data_confirms_acquisition( tmp_path: Path, ) -> None: service, runtime = service_with_fake_runtime(tmp_path) prepared = service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", compatibility_attestation=ATTESTATION, ) ) acquisition_id = prepared["acquisition"]["acquisition_id"] service.start_acquisition(StartAcquisitionRequest(acquisition_id=acquisition_id)) with pytest.raises(ValueError, match="подтверждённого потока point cloud"): service.stop_acquisition( StopAcquisitionRequest(acquisition_id=acquisition_id, mode="graceful") ) state = service.state() assert runtime.stop_calls == 0 assert state["acquisition"]["state"] == "starting" assert state["last_operation"]["action"] == "acquisition.start" assert state["last_operation"]["status"] == "running" @pytest.mark.parametrize("evidence_policy", ["best-effort", "disabled"]) def test_prepare_rejects_unsupported_evidence_policies( tmp_path: Path, evidence_policy: str, ) -> None: service, _ = service_with_fake_runtime(tmp_path) with pytest.raises(ValueError, match="evidence_policy=required"): service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", evidence_policy=evidence_policy, # type: ignore[arg-type] compatibility_attestation=ATTESTATION, ) ) assert service.state()["compatibility"]["profile_id"] is None @pytest.mark.parametrize( "requested_streams", [ DEFAULT_LIVE_STREAMS[:-1], (*DEFAULT_LIVE_STREAMS, DEFAULT_LIVE_STREAMS[0]), ], ) def test_prepare_rejects_stream_subsets_and_duplicates( tmp_path: Path, requested_streams: tuple[str, ...], ) -> None: service, _ = service_with_fake_runtime(tmp_path) with pytest.raises(ValueError, match="полный проверенный набор"): service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", requested_streams=requested_streams, # type: ignore[arg-type] compatibility_attestation=ATTESTATION, ) ) def test_exact_profile_is_inactive_until_explicit_operator_attestation( tmp_path: Path, ) -> None: service, _ = service_with_fake_runtime(tmp_path) initial = service.state() assert initial["compatibility"] == { "profile_id": None, "decision": "unknown", "permitted_mode": "evidence-only", "firmware_claim": "exact-3.0.2-profile-not-attested", "attestation": None, "vendor_writes_enabled": False, "camera_preview": "unverified", } assert initial["device_calibration"]["compatibility_profile_id"] is None attested = service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", compatibility_attestation=ATTESTATION, ) ) assert attested["compatibility"]["profile_id"] == XGRIDS_K1_COMPATIBILITY_PROFILE_ID assert attested["compatibility"]["decision"] == "limited" assert attested["compatibility"]["attestation"]["basis"] == "operator-attested" assert attested["device_session"]["compatibility_profile_id"] == ( XGRIDS_K1_COMPATIBILITY_PROFILE_ID ) def test_prepare_rejects_device_ap_fallback_as_direct_lan_target(tmp_path: Path) -> None: service, _ = service_with_fake_runtime(tmp_path) with pytest.raises(ValueError, match="точки доступа"): service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.56.1", compatibility_attestation=ATTESTATION, ) ) state = service.state() assert state["device_ref"] is None assert state["compatibility"]["profile_id"] is None def test_runtime_error_terminalizes_pending_graceful_stop(tmp_path: Path) -> None: service, runtime = service_with_fake_runtime(tmp_path) prepared = service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", compatibility_attestation=ATTESTATION, ) ) acquisition_id = prepared["acquisition"]["acquisition_id"] service.start_acquisition(StartAcquisitionRequest(acquisition_id=acquisition_id)) runtime.mark_ready() service.state() runtime.pcl_frames = 1 service.state() awaiting = service.stop_acquisition( StopAcquisitionRequest(acquisition_id=acquisition_id, mode="graceful") ) stop_operation_id = awaiting["last_operation"]["operation_id"] runtime.phase = "error" failed = service.state() stop_operation = next( item for item in failed["operations"] if item["operation_id"] == stop_operation_id ) assert failed["acquisition"]["state"] == "failed" assert stop_operation["status"] == "failed" assert stop_operation["error"]["side_effect_status"] == "unknown" def test_receiver_completion_terminalizes_unconfirmed_graceful_stop( tmp_path: Path, ) -> None: service, runtime = service_with_fake_runtime(tmp_path) prepared = service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", compatibility_attestation=ATTESTATION, ) ) acquisition_id = prepared["acquisition"]["acquisition_id"] service.start_acquisition(StartAcquisitionRequest(acquisition_id=acquisition_id)) runtime.mark_ready() service.state() runtime.pcl_frames = 1 service.state() awaiting = service.stop_acquisition( StopAcquisitionRequest(acquisition_id=acquisition_id, mode="graceful") ) stop_operation_id = awaiting["last_operation"]["operation_id"] runtime.phase = "idle" runtime.source_mode = "idle" runtime.source_ready = False failed = service.state() stop_operation = next( item for item in failed["operations"] if item["operation_id"] == stop_operation_id ) assert failed["acquisition"]["state"] == "failed" assert failed["acquisition"]["result"] == { "receiver_stopped": True, "device_state": "unknown", } assert stop_operation["status"] == "failed" assert stop_operation["error"]["code"] == ("receiver-completed-before-device-stop-confirmation") def test_unconfirmed_stop_operation_terminalizes_before_clock_seal_error( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: service, runtime = service_with_fake_runtime(tmp_path) prepared = service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", compatibility_attestation=ATTESTATION, ) ) acquisition_id = prepared["acquisition"]["acquisition_id"] service.start_acquisition(StartAcquisitionRequest(acquisition_id=acquisition_id)) runtime.pcl_frames = 1 service.state() awaiting = service.stop_acquisition( StopAcquisitionRequest(acquisition_id=acquisition_id, mode="graceful") ) stop_operation_id = awaiting["last_operation"]["operation_id"] out_dir = service._acquisition_out_dir # noqa: SLF001 assert out_dir is not None out_dir.mkdir() monkeypatch.setattr( facade_module, "seal_capture_clock", lambda _capture_root: (_ for _ in ()).throw(RuntimeError("synthetic seal failure")), ) runtime.phase = "idle" runtime.source_mode = "idle" with pytest.raises(RuntimeError, match="synthetic seal failure"): service.state() state = service.state() stop_operation = next( item for item in state["operations"] if item["operation_id"] == stop_operation_id ) assert state["acquisition"]["state"] == "failed" assert stop_operation["status"] == "failed" assert service._acquisition_session_lease is not None # noqa: SLF001 monkeypatch.setattr(facade_module, "seal_capture_clock", lambda _capture_root: None) service.stop() def test_abort_failure_terminalizes_acquisition_and_pending_start(tmp_path: Path) -> None: service, runtime = service_with_fake_runtime(tmp_path) prepared = service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", compatibility_attestation=ATTESTATION, ) ) acquisition_id = prepared["acquisition"]["acquisition_id"] service.start_acquisition(StartAcquisitionRequest(acquisition_id=acquisition_id)) runtime.stop_error = RuntimeError("synthetic receiver stop failure") with pytest.raises(RuntimeError, match="synthetic receiver stop failure"): service.abort_acquisition(AbortAcquisitionRequest(acquisition_id=acquisition_id)) state = service.state() operations = {item["action"]: item for item in state["operations"]} assert state["acquisition"]["state"] == "failed" assert state["acquisition"]["result"] == { "receiver_stopped": False, "device_state": "unknown", } assert operations["acquisition.start"]["status"] == "cancelled" assert operations["acquisition.abort"]["status"] == "failed" assert state["source_mode"] == "live" def test_abort_reserves_stopping_before_reentrant_runtime_callback( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: service, runtime = service_with_fake_runtime(tmp_path) prepared = service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", compatibility_attestation=ATTESTATION, ) ) acquisition_id = prepared["acquisition"]["acquisition_id"] service.start_acquisition(StartAcquisitionRequest(acquisition_id=acquisition_id)) runtime.pcl_frames = 1 service.state() out_dir = service._acquisition_out_dir # noqa: SLF001 assert out_dir is not None out_dir.mkdir() events: list[str] = [] monkeypatch.setattr( service.camera_preview, "stop_recording", lambda **_kwargs: events.append("camera"), ) def reentrant_runtime_stop() -> None: events.append("runtime") runtime.phase = "idle" runtime.source_mode = "idle" nested = service.state() assert nested["acquisition"]["state"] == "stopping" monkeypatch.setattr(runtime, "stop", reentrant_runtime_stop) monkeypatch.setattr( facade_module, "seal_capture_clock", lambda _capture_root: events.append("seal"), ) aborted = service.abort_acquisition(AbortAcquisitionRequest(acquisition_id=acquisition_id)) assert aborted["acquisition"]["state"] == "aborted" assert events == ["camera", "runtime", "seal"] assert service._acquisition_session_lease is None # noqa: SLF001 def test_clean_close_seals_and_cancels_pending_start_operation( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: service, runtime = service_with_fake_runtime(tmp_path) prepared = service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", compatibility_attestation=ATTESTATION, ) ) acquisition_id = prepared["acquisition"]["acquisition_id"] service.start_acquisition(StartAcquisitionRequest(acquisition_id=acquisition_id)) out_dir = service._acquisition_out_dir # noqa: SLF001 assert out_dir is not None out_dir.mkdir() events: list[str] = [] monkeypatch.setattr( service.camera_preview, "close", lambda: events.append("camera"), ) def close_runtime() -> None: events.append("runtime") runtime.phase = "idle" runtime.source_mode = "idle" monkeypatch.setattr(runtime, "close", close_runtime) monkeypatch.setattr( facade_module, "seal_capture_clock", lambda _capture_root: events.append("seal"), ) service.close() state = service.state() start_operation = next( item for item in state["operations"] if item["action"] == "acquisition.start" ) assert state["acquisition"]["state"] == "interrupted" assert start_operation["status"] == "cancelled" assert events == ["camera", "runtime", "seal"] assert service._acquisition_session_lease is None # noqa: SLF001 def test_clean_close_cancels_pending_external_stop_operation( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: service, runtime = service_with_fake_runtime(tmp_path) prepared = service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", compatibility_attestation=ATTESTATION, ) ) acquisition_id = prepared["acquisition"]["acquisition_id"] service.start_acquisition(StartAcquisitionRequest(acquisition_id=acquisition_id)) runtime.pcl_frames = 1 service.state() awaiting = service.stop_acquisition( StopAcquisitionRequest(acquisition_id=acquisition_id, mode="graceful") ) stop_operation_id = awaiting["last_operation"]["operation_id"] monkeypatch.setattr(service.camera_preview, "close", lambda: None) monkeypatch.setattr(runtime, "close", lambda: runtime.stop()) service.close() state = service.state() stop_operation = next( item for item in state["operations"] if item["operation_id"] == stop_operation_id ) assert state["acquisition"]["state"] == "interrupted" assert stop_operation["status"] == "cancelled" assert all( item["status"] not in {"accepted", "running", "operator_action_required"} for item in state["operations"] if item["action"].startswith("acquisition.") ) def test_close_failure_fails_pending_operation_and_retains_lease( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: service, runtime = service_with_fake_runtime(tmp_path) prepared = service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", compatibility_attestation=ATTESTATION, ) ) acquisition_id = prepared["acquisition"]["acquisition_id"] service.start_acquisition(StartAcquisitionRequest(acquisition_id=acquisition_id)) monkeypatch.setattr(service.camera_preview, "close", lambda: None) monkeypatch.setattr( runtime, "close", lambda: (_ for _ in ()).throw(RuntimeError("synthetic close timeout")), ) with pytest.raises(RuntimeError, match="synthetic close timeout"): service.close() state = service.state() start_operation = next( item for item in state["operations"] if item["action"] == "acquisition.start" ) assert state["acquisition"]["state"] == "failed" assert start_operation["status"] == "failed" assert start_operation["error"]["side_effect_status"] == "unknown" assert service._acquisition_session_lease is not None # noqa: SLF001 runtime.stop_error = None service.stop() assert service._acquisition_session_lease is None # noqa: SLF001 def test_replay_is_rejected_during_nonterminal_acquisition(tmp_path: Path) -> None: service, _ = service_with_fake_runtime(tmp_path) service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", compatibility_attestation=ATTESTATION, ) ) with pytest.raises(RuntimeError, match="активной acquisition-сессии"): service.start_replay("sessions/fixture.k1mqtt", speed=1.0, loop=False) assert service.state()["acquisition"]["state"] == "prepared" def test_prepare_rejects_active_replay_without_stopping_or_replacing_it( tmp_path: Path, ) -> None: service, runtime = service_with_fake_runtime(tmp_path) runtime.phase = "replay" runtime.source_mode = "replay" runtime.source_ready = True with pytest.raises(RuntimeError, match="активного live/replay"): service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", compatibility_attestation=ATTESTATION, ) ) state = service.state() assert state["acquisition"] is None assert state["source_mode"] == "replay" assert runtime.stop_calls == 0 assert service._acquisition_session_lease is None # noqa: SLF001 def test_network_provisioning_is_single_flight_and_secret_is_unwrapped_only_at_boundary( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: service, _ = service_with_fake_runtime(tmp_path) service._devices = [ # noqa: SLF001 - deliberate white-box concurrency fixture {"device_id": "k1-a"}, {"device_id": "k1-b"}, ] boundary_calls: list[tuple[str, str, str]] = [] async def scenario() -> dict[str, Any]: entered = asyncio.Event() release = asyncio.Event() async def fake_provision( device_id: str, ssid: str, password: str, **_: object, ) -> dict[str, Any]: boundary_calls.append((device_id, ssid, password)) entered.set() await release.wait() return { "started_at_utc": "2026-07-16T12:00:00Z", "completed_at_utc": "2026-07-16T12:00:01Z", "profile_id": "xgrids-k1-fw3-wifi-v1", "outcome": "lan_address_observed", "observations": [{"status": {"ipv4": "192.168.1.20"}}], } monkeypatch.setattr(facade_module, "provision_wifi_once", fake_provision) first = asyncio.create_task( service.connect( ConnectRequest( device_id="k1-a", ssid="lab-network", password=SecretStr(PRIMARY_TEST_CREDENTIAL), compatibility_attestation=ATTESTATION, ) ) ) await asyncio.wait_for(entered.wait(), timeout=1.0) with pytest.raises(RuntimeError, match="уже выполняется"): await service.connect( ConnectRequest( device_id="k1-b", ssid="other-network", password=SecretStr(SECONDARY_TEST_CREDENTIAL), compatibility_attestation=ATTESTATION, ) ) release.set() return await asyncio.wait_for(first, timeout=1.0) connected = asyncio.run(scenario()) assert boundary_calls == [("k1-a", "lab-network", PRIMARY_TEST_CREDENTIAL)] assert connected["k1_ip"] == "192.168.1.20" assert PRIMARY_TEST_CREDENTIAL not in str(connected) provision_operations = [ item for item in connected["operations"] if item["action"] == "network.provision" ] assert {item["status"] for item in provision_operations} == {"succeeded", "failed"} def test_provisioning_cannot_switch_device_during_active_acquisition( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: service, _ = service_with_fake_runtime(tmp_path) service._devices = [{"device_id": "k1-a"}] # noqa: SLF001 service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", compatibility_attestation=ATTESTATION, ) ) called = False async def should_not_run(*_: object, **__: object) -> dict[str, Any]: nonlocal called called = True raise AssertionError("provisioning boundary must not be reached") monkeypatch.setattr(facade_module, "provision_wifi_once", should_not_run) with pytest.raises(RuntimeError, match="активной acquisition-сессии"): asyncio.run( service.connect( ConnectRequest( device_id="k1-a", ssid="lab-network", password=SecretStr(PRIMARY_TEST_CREDENTIAL), compatibility_attestation=ATTESTATION, ) ) ) assert called is False def test_sensor_catalog_exposes_two_browser_adapter_cameras_after_profile_attestation( tmp_path: Path, ) -> None: service, _ = service_with_fake_runtime(tmp_path) initial = service.state() initial_cameras = [ stream for stream in initial["sensor_catalog"]["streams"] if stream.get("semantic_channel_id") == "camera.preview.live" ] assert {stream["source_id"] for stream in initial_cameras} == { "sensor.camera.left", "sensor.camera.right", } assert all(stream["availability"] == "unverified" for stream in initial_cameras) state = service.prepare_acquisition( PrepareAcquisitionRequest( project_name=PROJECT_NAME, host="192.168.1.20", compatibility_attestation=ATTESTATION, ) ) cameras = [ stream for stream in state["sensor_catalog"]["streams"] if stream.get("semantic_channel_id") == "camera.preview.live" ] assert len(cameras) == 2 assert all(camera["availability"] == "available" for camera in cameras) assert all(camera["modality"] == "encoded-video" for camera in cameras) assert all(camera["decode_status"] == "rtsp-h264-observed-browser-remux" for camera in cameras) assert all(camera["activation"]["max_active"] == 1 for camera in cameras) assert all(camera["delivery"] is None for camera in cameras) assert state["connection_verification"]["network_reachability"] == "unknown" assert state["device_calibration"]["status"] == "unavailable" assert state["device_calibration"]["vehicle_extrinsics"] == ("host-domain-not-owned-by-plugin")