import asyncio import copy import json import secrets from datetime import UTC, datetime, timedelta from pathlib import Path import pytest from missioncore_plugin_sdk.v0alpha2.session import DeviceSessionSnapshot from k1link.device_plugins.xgrids_k1.linux_host import nm_fields, route_fields from k1link.device_plugins.xgrids_k1.node_bridge import NodeBridge, plugin_operation_id from k1link.device_plugins.xgrids_k1.node_sensor import NodeK1Sensor, project_sensor from k1link.viewer.node_rerun import NodeRerunHub def state(): return { "snapshot_runtime_id": "runtime-one", "snapshot_revision": 1, "devices": [ {"device_id": "AA:BB:CC:DD:EE:FF", "name": "K1-test", "likely_k1": True}, {"device_id": "unrelated", "name": "Other", "likely_k1": False}, ], "ble_discovery_generation": 1, "desired_connection_mode_revision": 0, "active_connection_mode": "bridge", "connection_lifecycle": {"connection_ready": True, "ready_to_start": True}, "selected_device_id": "AA:BB:CC:DD:EE:FF", "source_mode": "idle", "device_session": { "device_id": "synthetic-k1", "device_session_id": "session-test", "opened_at": datetime.now(UTC).isoformat(), }, "application_control_session": { "session_generation": 1, "state_revision": 1, "state": "connection-ready", }, } class Facade: def __init__(self): self.current = state() self.actions = [] async def invoke(self, request): self.actions.append((request.action_id, copy.deepcopy(request.parameters))) if request.action_id == "application-control.workspace.enter": self.current["application_control_session"]["state"] = "workspace-ready" elif request.action_id == "acquisition.prepare": self.current["acquisition"] = { "acquisition_id": "acquisition-test", "state": "prepared", "state_revision": 1, } self.current["application_control_session"]["state"] = "project-ready" elif request.action_id == "acquisition.start": self.current["application_control_session"]["state"] = "initializing" self.current["acquisition"]["state"] = "running" self.current["source_mode"] = "live" return copy.deepcopy(self.current) def bridge(): result = NodeBridge(Path.cwd(), service=object()) result.facade = Facade() return result def test_journalled_failure_logs_source_without_secret_and_does_not_repeat(caplog): from bleak.exc import BleakError secret = secrets.token_hex(20) device = bridge() calls = [] identifier = "op_" + "c" * 32 native = plugin_operation_id(identifier) async def invoke(action, payload, _identifier): calls.append(action) if action == "network.provision": raise BleakError(secret) return {"operations": [{"operation_id": native, "status": "failed"}]} device.invoke = invoke result = asyncio.run(device.invoke_journalled( "network.provision", {"operation_id": native, "password": secret}, identifier, )) assert result["operations"][0]["status"] == "failed" assert calls == ["network.provision", "state.read"] assert "exception=BleakError" in caplog.text assert "test_node_k1_bridge.py:invoke:" in caplog.text assert identifier in caplog.text assert secret not in caplog.text def test_real_facade_preserves_native_failure_location_in_journalled_log(caplog): from bleak.exc import BleakError secret = secrets.token_hex(20) calls = [] class Service: current = state() def bind_runtime_event_loop(self, _loop): pass def require_snapshot_runtime_id(self, expected): assert expected == self.current["snapshot_runtime_id"] def state(self): calls.append("state") return copy.deepcopy(self.current) async def connect(self, request): calls.append("connect") self.current["operations"] = [{ "operation_id": request.operation_id, "status": "failed", "error": {"code": "BleakError"}, }] raise BleakError(secret) async def run(): device = NodeBridge(Path.cwd(), service=Service()) command = { "operation_id": "op_" + "c" * 32, "runtime_id": "runtime-one", "deadline_at": (datetime.now(UTC) + timedelta(seconds=60)).isoformat(), "action": "connect", "parameters": { "device_id": "AA:BB:CC:DD:EE:FF", "discovery_generation": 1, "mode_revision": 0, "ssid": "test-network", "password": secret, }, } result = await device.deliver(command) assert result["command_result"]["status"] == "failed" assert calls == ["state", "connect", "state"] assert "password" not in command["parameters"] assert secret not in json.dumps(result) asyncio.run(run()) assert "PluginExecutionError[" in caplog.text assert "BleakError[" in caplog.text assert "test_node_k1_bridge.py:connect:" in caplog.text assert secret not in caplog.text assert "test-network" not in caplog.text def test_failure_locations_respects_suppression_and_bounds_cyclic_chains(): from k1link.device_plugins.xgrids_k1.node_bridge import failure_locations first = RuntimeError(secrets.token_hex(20)) second = ValueError(secrets.token_hex(20)) first.__context__ = second second.__cause__ = first assert failure_locations(first) == "RuntimeError[] <- ValueError[]" first.__suppress_context__ = True assert failure_locations(first) == "RuntimeError[]" def test_node_scan_reaches_service_through_real_facade_with_runtime_fence(): """Exercise the actual admission boundary, replacing only the BLE service.""" from k1link.device_plugins.xgrids_k1.facade import SnapshotRuntimeConflict class Service: def __init__(self): self.scans = [] self.fences = [] self.current = state() def bind_runtime_event_loop(self, _loop): pass def state(self): return copy.deepcopy(self.current) def require_snapshot_runtime_id(self, expected): if expected != self.current["snapshot_runtime_id"]: raise SnapshotRuntimeConflict() self.fences.append(expected) async def scan_ble(self, request): self.scans.append(request.operation_id) return self.state() async def run(): service = Service() device = NodeBridge(Path.cwd(), service=service) command = { "operation_id": "op_" + "b" * 32, "runtime_id": "runtime-one", "deadline_at": (datetime.now(UTC) + timedelta(seconds=60)).isoformat(), "action": "scan", "parameters": {}, } output = await device.deliver(command) assert service.fences == ["runtime-one"] assert service.scans == [plugin_operation_id(command["operation_id"])] assert len(output["candidates"]) == 1 command["runtime_id"] = "retired-runtime" rejected = await device.deliver(command) assert rejected["command_result"]["status"] == "rejected" assert service.scans == [plugin_operation_id(command["operation_id"])] asyncio.run(run()) @pytest.mark.parametrize("radio_failure", [False, True]) def test_node_scan_real_service_preserves_operation_identity_and_replay( tmp_path, monkeypatch, radio_failure, ): from bleak.backends.device import BLEDevice from bleak.backends.scanner import AdvertisementData from k1link.device_plugins.xgrids_k1 import facade from k1link.device_plugins.xgrids_k1.ble import scanner calls = [] class Radio: def __init__(self, detection_callback): self.callback = detection_callback async def __aenter__(self): calls.append("scan") if radio_failure: raise RuntimeError("synthetic radio unavailable") self.callback( BLEDevice("AA:BB:CC:DD:EE:FF", "XGR-TEST", None), AdvertisementData( local_name="XGR-TEST", manufacturer_data={}, service_data={}, service_uuids=[], tx_power=None, rssi=-50, platform_data=(), ), ) return self async def __aexit__(self, *_args): pass # Keep the actual Node adapter, facade, service, journal and BLE arbiter. # Replace only the OS radio endpoint and shorten its observation window. monkeypatch.setattr(scanner, "BleakScanner", Radio) monkeypatch.setattr(scanner, "BLE_SCAN_INITIAL_WINDOW_SECONDS", 0.01) service = facade.XgridsK1CompatibilityService(tmp_path) async def run(): device = NodeBridge(tmp_path, service=service) current = await device.state() command = { "operation_id": "op_" + "b" * 32, "runtime_id": current["runtime_id"], "deadline_at": (datetime.now(UTC) + timedelta(seconds=60)).isoformat(), "action": "scan", "parameters": {}, } result = await device.deliver(command) if not radio_failure: assert scanner.capture_discovered_device("AA:BB:CC:DD:EE:FF") is not None repeated = await device.deliver(command) assert len(calls) == 1 assert result["discovery_generation"] == 1 if radio_failure: assert result["candidates"] == [] else: assert result["candidates"][0]["id"] == "AA:BB:CC:DD:EE:FF" assert result["command_result"] == repeated["command_result"] == { "operation_id": command["operation_id"], "action": "scan", "status": "failed" if radio_failure else "succeeded", "error_code": "RuntimeError" if radio_failure else None, } try: asyncio.run(run()) finally: service.close() def test_node_attempt_projection_keeps_host_correlation_without_mutating_plugin_state(): current = state() identifier = "op_" + "a" * 32 current["connection_attempt"] = { "schema_version": "missioncore.xgrids-k1-connection-attempt/v1", "attempt_id": plugin_operation_id(identifier), "recovery_operation_id": plugin_operation_id("op_" + "b" * 32), "status": "failed", "public_error_code": "wifi-ssid-not-found", } result = NodeBridge.project(current) assert result["connection_attempt"]["attempt_id"] == identifier assert result["connection_attempt"]["recovery_operation_id"] == "op_" + "b" * 32 assert current["connection_attempt"]["attempt_id"] == plugin_operation_id(identifier) def test_node_bridge_forces_reviewed_bridge_and_clears_input_secret(): async def run(): device = bridge() command = { "operation_id": "op_" + "a" * 32, "runtime_id": "runtime-one", "deadline_at": (datetime.now(UTC) + timedelta(seconds=60)).isoformat(), "action": "connect", "parameters": { "device_id": "AA:BB:CC:DD:EE:FF", "discovery_generation": 1, "mode_revision": 0, "ssid": "test-network", "password": secrets.token_urlsafe(24), "allow_host_wifi_switch": True, }, } secret = command["parameters"]["password"] output = await device.execute(command) payload = next(v for a, v in device.facade.actions if a == "network.provision") assert payload["connection_mode"] == "bridge" assert payload["allow_host_wifi_switch"] is False assert payload["password"] == secret assert "password" not in command["parameters"] assert secret not in json.dumps(output) assert [v["id"] for v in output["candidates"]] == ["AA:BB:CC:DD:EE:FF"] asyncio.run(run()) def test_stale_discovery_prevents_provisioning(): async def run(): device = bridge() with pytest.raises(ValueError): await device.execute( { "operation_id": "op_" + "a" * 32, "runtime_id": "runtime-one", "deadline_at": (datetime.now(UTC) + timedelta(seconds=60)).isoformat(), "action": "connect", "parameters": { "device_id": "AA:BB:CC:DD:EE:FF", "discovery_generation": 0, "mode_revision": 0, }, } ) assert all(action != "network.provision" for action, _ in device.facade.actions) asyncio.run(run()) def test_sensor_projection_binds_native_sdk_to_board(): value = project_sensor(state(), "node-test") snapshot = DeviceSessionSnapshot.model_validate(value["snapshot"]) assert snapshot.context.execution.node_id == "node-test" assert snapshot.context.device.device_id == value["id"] assert value["kind"] == "k1" def test_one_start_intent_preserves_canonical_enter_prepare_start_sequence(): async def run(): device = bridge() sensor = NodeK1Sensor(device, None) item = project_sensor(state(), "node-test") command = { "operation_id": "op_" + "a" * 32, "action_id": "start", "session": {"device_id": item["id"], "session_id": "session-test"}, "deadline_at": (datetime.now(UTC) + timedelta(seconds=60)).isoformat(), "parameters": { "operator_confirmed": True, "control_generation": 1, "acquisition_id": None, }, } result = await sensor.execute(command, "node-test") actions = [a for a, _ in device.facade.actions if a != "state.read"] assert actions == [ "application-control.workspace.enter", "acquisition.prepare", "acquisition.start", ] from k1link.web.device_lifecycle import OperationJournal journal = OperationJournal() for action, payload in device.facade.actions: if action in {"acquisition.prepare", "acquisition.start"}: row, created = journal.begin( action, operation_id=payload["operation_id"], idempotency_key=payload["idempotency_key"], ) assert created assert journal.begin(action, operation_id=row.operation_id)[1] is False assert result["snapshot"]["acquisition"] == "streaming" command["parameters"]["control_generation"] = 2 with pytest.raises(ValueError): await sensor.execute(command, "node-test") assert len([a for a, _ in device.facade.actions if a == "acquisition.start"]) == 1 asyncio.run(run()) def test_networkmanager_ssids_are_not_split_at_escaped_colons(): assert nm_fields(r"field\:network\\name:88:WPA2") == ["field:network\\name", "88", "WPA2"] def test_wired_board_bridge_does_not_require_host_wifi_association(tmp_path, monkeypatch): from k1link.device_plugins.xgrids_k1 import linux_host (tmp_path / "enp1s0").mkdir() def forbidden(*_args, **_kwargs): raise AssertionError("Ethernet must not request Wi-Fi association") monkeypatch.setattr(linux_host, "_run", forbidden) value = linux_host.LinuxWifiAssociationProbe(tmp_path).observe(interface_name="enp1s0") assert value["association_state"] == "not-wifi" assert value["continuity_proven"] is True assert value["reason_code"] is None @pytest.mark.parametrize("route_changes", [False, True]) def test_wired_board_real_service_correlates_route_and_control_endpoint( tmp_path, monkeypatch, route_changes, ): from dataclasses import replace from k1link.device_plugins.xgrids_k1 import facade, linux_host sys_net = tmp_path / "sys-net" (sys_net / "enp1s0").mkdir(parents=True) probe = linux_host.LinuxWifiAssociationProbe(sys_net) service = facade.XgridsK1CompatibilityService( tmp_path / "runtime", host_wifi_association_probe=probe, ) path = facade.HostPathProbeResult( available=True, fingerprint="synthetic-route", interface="enp1s0", source_ipv4="192.168.1.2", route_class="direct", reason_code=None, ) monkeypatch.setattr(facade, "_inspect_host_path", lambda _target: path) calls = [] def tcp(_target): calls.append("tcp") if route_changes: monkeypatch.setattr(facade, "_inspect_host_path", lambda _target: replace( path, fingerprint="new-route", source_ipv4="192.168.1.3", )) return True monkeypatch.setattr(facade, "_control_endpoint_reachable", tcp) def forbidden(*_args, **_kwargs): raise AssertionError("Wired control bootstrap must not request host Wi-Fi") monkeypatch.setattr(linux_host, "_run", forbidden) try: observed = service._sample_host_path("192.168.1.7") assert observed.available is True assert observed.interface == "enp1s0" assert observed.reason_code is None assert observed.fingerprint != path.fingerprint assert service._sample_host_path("192.168.1.7").fingerprint == observed.fingerprint endpoint = service._probe_control_endpoint("192.168.1.7") assert calls == ["tcp"] assert endpoint.reachable is (not route_changes) assert endpoint.reason_code == ( "host-path-changed-during-tcp-probe" if route_changes else None ) finally: service.close() def test_linux_kernel_route_is_matched_route_not_resolved_host(monkeypatch): from k1link.device_plugins.xgrids_k1 import linux_host from k1link.device_plugins.xgrids_k1.facade import _classify_host_route calls = [] def run(args): calls.append(args) return '[{"dst":"default","dev":"wlp2s0","gateway":"192.168.1.1"}]' monkeypatch.setattr(linux_host, "_run", run) result = route_fields("192.168.2.7") assert calls[0] == ["ip", "-j", "route", "get", "192.168.2.7", "fibmatch"] assert _classify_host_route(result["interface"], result["destination"])[0] == "default" assert _classify_host_route("wlp2s0", "192.168.1.0/24")[0] == "direct" assert _classify_host_route("tailscale0", "192.168.1.7")[0] == "tunnel" def test_native_node_rrd_opens_no_grpc_listener(monkeypatch): from k1link.viewer import rerun_bridge def forbidden(*_args, **_kwargs): raise AssertionError("Node must not expose a gRPC listener") monkeypatch.setattr(rerun_bridge, "_select_available_grpc_port", forbidden) hub = NodeRerunHub() publisher = hub.create() publisher.begin_session() subscriber = hub.subscribe() try: data = subscriber.read() assert data[:4] == b"RRF2" finally: subscriber.close() subscriber.thread.join(6) publisher.close()