from __future__ import annotations import copy import importlib.util from pathlib import Path from types import ModuleType from typing import Any import pytest REPOSITORY_ROOT = Path(__file__).parents[1] LOADER_PATH = REPOSITORY_ROOT / "plugins" / "xgrids-k1" / "profile_loader.py" def _load_module() -> ModuleType: spec = importlib.util.spec_from_file_location("xgrids_k1_profile_loader", LOADER_PATH) assert spec is not None assert spec.loader is not None module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module LOADER = _load_module() def _by_id(items: list[dict[str, Any]]) -> dict[str, dict[str, Any]]: return {item["id"]: item for item in items} def _boolean_write_flags(value: Any) -> list[bool]: flags: list[bool] = [] if isinstance(value, dict): write_enabled = value.get("write_enabled") if isinstance(write_enabled, bool): flags.append(write_enabled) for child in value.values(): flags.extend(_boolean_write_flags(child)) elif isinstance(value, list): for child in value: flags.extend(_boolean_write_flags(child)) return flags def test_xgrids_compatibility_profile_loads_exact_firmware_and_sources() -> None: profile = LOADER.load_compatibility_profile() assert profile["profile_id"] == "xgrids.lixelkity-k1.fw-3.0.2.direct-lan.v1" assert profile["scope"]["firmware"] == {"match": "exact", "version": "3.0.2"} assert profile["scope"]["topology"] == "direct-lan" assert LOADER.matches_target(profile, firmware="3.0.2", topology="direct-lan") assert not LOADER.matches_target(profile, firmware="3.0.3", topology="direct-lan") assert not LOADER.matches_target(profile, firmware="3.0.2", topology="device-ap") for source in profile["evidence_sources"]: assert (REPOSITORY_ROOT / source["path"]).is_file() def test_xgrids_compatibility_profile_keeps_evidence_levels_independent() -> None: profile = LOADER.load_compatibility_profile() channels = _by_id(profile["channels"]) verified_stream = { "observed": True, "decoded": True, "replay_verified": True, "physical_verified": True, "write_enabled": False, } raw_status = { "observed": True, "decoded": False, "replay_verified": False, "physical_verified": True, "write_enabled": False, } observed_raw = { "observed": True, "decoded": False, "replay_verified": False, "physical_verified": True, "write_enabled": False, } assert channels["spatial.point-cloud.live"]["evidence"] == verified_stream assert channels["spatial.pose.live"]["evidence"] == verified_stream assert channels["device.status.live"]["evidence"] == raw_status assert channels["device.heartbeat.live"]["evidence"] == raw_status assert channels["device.status.live"]["semantic_payload"] is None camera = channels["camera.preview.live"] assert camera["discovery_status"] == "observed" assert camera["topic"] is None assert set(camera["endpoint_templates"]) == { "rtsp://{confirmed-device-private-ipv4}:8554/live/chn_left_main", "rtsp://{confirmed-device-private-ipv4}:8554/live/chn_right_main", } assert "H.264" in camera["wire_format"] assert camera["evidence"] == observed_raw def test_xgrids_compatibility_profile_declares_only_reviewed_transports() -> None: profile = LOADER.load_compatibility_profile() transports = _by_id(profile["transports"]) ble = transports["ble.wifi-bootstrap.fw3.v1"] assert ble["service_uuid"] == "00007f00-0000-1000-8000-00805f9b34fb" assert ble["characteristics"] == { "wifi_request": "00007f01-0000-1000-8000-00805f9b34fb", "wifi_status": "00007f02-0000-1000-8000-00805f9b34fb", } assert ble["request_frame_bytes"] == 99 assert ble["evidence"]["observed"] is True assert ble["evidence"]["physical_verified"] is True assert ble["evidence"]["write_enabled"] is False mqtt = transports["mqtt.direct-lan.fw3.v1"] assert mqtt["protocol"] == "MQTT 3.1.1" assert mqtt["network"]["transport"] == "TCP" assert mqtt["network"]["port"] == 1883 assert mqtt["network"]["tls"] is False assert "lixel/application/report/#" in mqtt["subscription_allowlist"] assert not any("/request/" in topic for topic in mqtt["subscription_allowlist"]) assert mqtt["evidence"]["observed"] is True assert mqtt["evidence"]["decoded"] is False assert mqtt["evidence"]["write_enabled"] is False rtsp = transports["rtsp.camera-preview.fw3.v1"] assert rtsp["protocol"] == "RTSP 1.0 with interleaved RTP over TCP" assert rtsp["network"]["port"] == 8554 assert rtsp["media"]["codec"] == "H.264" assert rtsp["media"]["rtp_payload_type"] == 96 assert rtsp["evidence"]["observed"] is True assert rtsp["evidence"]["write_enabled"] is False def test_xgrids_compatibility_profile_maps_actions_without_enabling_writes() -> None: profile = LOADER.load_compatibility_profile() control = profile["acquisition_control"] actions = _by_id(control["semantic_actions"]) assert profile["safety"]["default_mode"] == "read-only" assert profile["safety"]["vendor_writes_enabled"] is False assert control["mode"] == "operator-manual" assert control["verified_device_control"]["gesture"] == "physical-double-click" for action_id, action_code in (("acquisition.start", 1), ("acquisition.stop", 2)): action = actions[action_id] mapping = action["vendor_request_mapping"] assert action["execution"] == "operator-manual" assert mapping["evidence_kind"] == "owner-controlled-wire-observation" assert mapping["topic"] == "lixel/application/request/modeling" assert mapping["qos"] == 2 assert mapping["message_type"] == "ModelingRequest" assert mapping["action_field_value"] == action_code assert mapping["required_unresolved_context"] assert mapping["evidence"]["observed"] is True assert mapping["evidence"]["decoded"] is True assert mapping["evidence"]["physical_verified"] is True assert mapping["evidence"]["replay_verified"] is False assert mapping["write_enabled"] is False calibration = actions["calibration.device.start"] assert calibration["execution"] == "unavailable" assert calibration["vendor_request_mapping"] is None assert calibration["evidence"]["observed"] is False assert _boolean_write_flags(profile) assert not any(_boolean_write_flags(profile)) def test_xgrids_compatibility_profile_rejects_vendor_write_promotion() -> None: profile = LOADER.load_compatibility_profile() modified = copy.deepcopy(profile) actions = _by_id(modified["acquisition_control"]["semantic_actions"]) actions["acquisition.start"]["vendor_request_mapping"]["write_enabled"] = True with pytest.raises(LOADER.CompatibilityProfileError, match="must remain false"): LOADER.validate_compatibility_profile(modified) def test_xgrids_compatibility_profile_rejects_claimed_camera_endpoint() -> None: profile = LOADER.load_compatibility_profile() modified = copy.deepcopy(profile) channels = _by_id(modified["channels"]) channels["camera.preview.live"]["endpoint_templates"].append("rtsp://unverified") with pytest.raises(LOADER.CompatibilityProfileError, match="camera endpoint templates"): LOADER.validate_compatibility_profile(modified) def test_xgrids_compatibility_profile_rejects_duplicate_json_keys(tmp_path: Path) -> None: profile_path = tmp_path / "duplicate.json" profile_path.write_text('{"schema_version": 1, "schema_version": 1}', encoding="utf-8") with pytest.raises(LOADER.CompatibilityProfileError, match="duplicate JSON key"): LOADER.load_compatibility_profile(profile_path)