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"]["vendor"] == "XGRIDS" assert profile["scope"]["model"] == "LixelKity K1" 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, } decoded_status = { "observed": True, "decoded": True, "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.modeling.live"]["evidence"] == decoded_status assert channels["device.status.live"]["evidence"] == decoded_status assert channels["device.heartbeat.live"]["evidence"] == observed_raw assert "ScanTime" in channels["device.modeling.live"]["semantic_payload"] assert "modeling-state" in channels["device.status.live"]["semantic_payload"] 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["retain"] is False assert mapping["message_type"] == "ModelingRequest" assert mapping["action_field_value"] == action_code assert mapping["header_contract"]["session_id"] == "{device_id}:ModelingRequest" assert mapping["success_result_code"] == 302_252_033 if action_id == "acquisition.start": assert mapping["request_fields"] == { "project_name": "required-operator-value", "record_mode": 2, "scan_mode": 1, "mount_type": 0, "pre_project_id": "omitted-in-retained-request", } else: assert mapping["request_fields"] == {} 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_noncanonical_modeling_header() -> None: profile = LOADER.load_compatibility_profile() modified = copy.deepcopy(profile) actions = _by_id(modified["acquisition_control"]["semantic_actions"]) actions["acquisition.start"]["vendor_request_mapping"]["header_contract"]["session_id"] = ( "caller-provided" ) with pytest.raises(LOADER.CompatibilityProfileError, match="header contract"): LOADER.validate_compatibility_profile(modified) def test_xgrids_compatibility_profile_pins_vendor_and_inert_request_mapping() -> None: profile = LOADER.load_compatibility_profile() wrong_model = copy.deepcopy(profile) wrong_model["scope"]["vendor"] = "OTHER" with pytest.raises(LOADER.CompatibilityProfileError, match="vendor/model"): LOADER.validate_compatibility_profile(wrong_model) wrong_message = copy.deepcopy(profile) wrong_actions = _by_id(wrong_message["acquisition_control"]["semantic_actions"]) wrong_actions["acquisition.start"]["vendor_request_mapping"]["message_type"] = "OtherRequest" with pytest.raises(LOADER.CompatibilityProfileError, match="message type"): LOADER.validate_compatibility_profile(wrong_message) missing_write_gate = copy.deepcopy(profile) missing_actions = _by_id(missing_write_gate["acquisition_control"]["semantic_actions"]) del missing_actions["acquisition.stop"]["vendor_request_mapping"]["write_enabled"] with pytest.raises(LOADER.CompatibilityProfileError, match="explicitly"): LOADER.validate_compatibility_profile(missing_write_gate) wrong_telemetry = copy.deepcopy(profile) wrong_channels = _by_id(wrong_telemetry["channels"]) wrong_channels["device.modeling.live"]["bounds"]["max_mqtt_payload_bytes"] = 1 with pytest.raises(LOADER.CompatibilityProfileError, match="telemetry channel"): LOADER.validate_compatibility_profile(wrong_telemetry) 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)