263 lines
11 KiB
Python
263 lines
11 KiB
Python
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"]["platform_type"] == "A4"
|
|
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["software_acceptance_transport"] == {
|
|
"status": "installed-operator-present",
|
|
"default_authority": "disabled",
|
|
"profile_gate": "live-device-info-exact-match",
|
|
"dialogue": "single-socket-canonical-start-to-stop",
|
|
"automatic_retry": False,
|
|
"supported_mount_type": "handheld",
|
|
"supported_gnss_mode": "none",
|
|
}
|
|
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)
|