363 lines
14 KiB
Python
363 lines
14 KiB
Python
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_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
|
|
|
|
|
|
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()
|