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 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_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", ] 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_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()