NODEDC_MISSION_CORE/tests/test_xgrids_application_acc...

307 lines
11 KiB
Python

from __future__ import annotations
import hashlib
from collections.abc import Collection, Sequence
import pytest
from k1link.device_plugins.xgrids_k1.protocol.application_acceptance import (
ApplicationAcceptanceError,
PhysicalAcceptanceChecklist,
PhysicalAcceptanceDialogueExecutor,
PhysicalAcceptancePermit,
)
from k1link.device_plugins.xgrids_k1.protocol.application_bootstrap import (
DEVICE_CONFIG_RESPONSE_TOPIC,
ApplicationControlAuthority,
ShadowApplicationBootstrapOrchestrator,
)
from k1link.device_plugins.xgrids_k1.protocol.application_mqtt import (
MODELING_RESPONSE_TOPIC,
)
from k1link.device_plugins.xgrids_k1.protocol.application_publish import (
OneShotPublishEnvelope,
)
from k1link.device_plugins.xgrids_k1.protocol.modeling_control import (
OPENAPI_SUCCESS,
CommandHeaderIdentity,
ModelingAction,
MountType,
RecordMode,
ScanMode,
encode_modeling_start,
encode_modeling_stop,
)
from k1link.device_plugins.xgrids_k1.protocol.modeling_safety import (
ShadowModelingCommand,
)
APPLICATION_KEY = "11111111-2222-3333-4444-555555555555"
VENDOR_DEVICE_ID = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"
def _varint(value: int) -> bytes:
encoded = bytearray()
while value > 0x7F:
encoded.append((value & 0x7F) | 0x80)
value >>= 7
encoded.append(value)
return bytes(encoded)
def _uint(number: int, value: int) -> bytes:
return _varint(number << 3) + _varint(value)
def _bytes(number: int, value: bytes) -> bytes:
return _varint((number << 3) | 2) + _varint(len(value)) + value
def _text(number: int, value: str) -> bytes:
return _bytes(number, value.encode())
def _header(session_id: str) -> bytes:
return _text(4, VENDOR_DEVICE_ID) + _text(5, session_id) + _text(6, APPLICATION_KEY)
def _device_info_response(session_id: str) -> bytes:
base_info = b"".join(
(
_text(2, "V3.0.2-20260101-release"),
_text(3, "V3.0.2"),
_text(6, "LixelKity K1"),
_text(7, "K1SERIAL01"),
_text(8, "K1"),
)
)
working_status = _uint(1, 1)
device_info = _bytes(2, base_info) + _bytes(7, working_status)
error = _uint(1, OPENAPI_SUCCESS)
return _bytes(1, _header(session_id)) + _bytes(2, device_info) + _bytes(15, error)
def _generic_response(session_id: str) -> bytes:
return _bytes(1, _header(session_id)) + _bytes(15, _uint(1, OPENAPI_SUCCESS))
def _modeling_response(action: ModelingAction) -> bytes:
session = f"{VENDOR_DEVICE_ID}:ModelingRequest"
return _bytes(1, _header(session)) + _uint(2, action) + _bytes(15, _uint(1, OPENAPI_SUCCESS))
class SyntheticAcceptanceTransport:
def __init__(self) -> None:
self.batches: list[tuple[str, ...]] = []
def exchange_batch_once(
self,
envelopes: Sequence[OneShotPublishEnvelope],
*,
required_response_topics: Collection[str],
) -> dict[str, bytes]:
self.batches.append(tuple(envelope.operation_key for envelope in envelopes))
responses: dict[str, bytes] = {}
if MODELING_RESPONSE_TOPIC in required_response_topics:
action = (
ModelingAction.START
if envelopes[0].operation_key == "modeling:start"
else ModelingAction.STOP
)
responses[MODELING_RESPONSE_TOPIC] = _modeling_response(action)
return responses
for envelope in envelopes:
prefix, ordinal_text, message_type = envelope.operation_key.split(":", 2)
assert prefix == "bootstrap"
ordinal = int(ordinal_text)
if message_type == "DeviceInfoRequest":
session = (
":DeviceInfoRequest"
if ordinal == 1
else f"{VENDOR_DEVICE_ID}:DeviceInfoRequest"
)
response = _device_info_response(session)
else:
session = (
f":{message_type}" if ordinal <= 3 else f"{VENDOR_DEVICE_ID}:{message_type}"
)
if message_type == "DeviceConfigRequest":
session += ":Publish_Proto_DeviceConfig_SetTime"
response = _generic_response(session)
for topic in required_response_topics:
if topic.endswith(_response_suffix(message_type)):
responses[topic] = response
return responses
def _response_suffix(message_type: str) -> str:
values = {
"DeviceInfoRequest": "device_info",
"ModelingStatusRequest": "modeling_status",
"GetRtkAdvanceRequest": "get_rtk_advance",
"DeviceConfigRequest": "device_config",
"GetNtripProfileRequest": "get_ntrip_profile",
"GetCloudServerConfigRequest": "get_cloud_server_config",
}
return values[message_type]
def _checklist(action: ModelingAction) -> PhysicalAcceptanceChecklist:
return PhysicalAcceptanceChecklist(
action=action,
operator_present=True,
owner_controlled_device=True,
lixelgo_closed=True,
battery_storage_confirmed=True,
expected_physical_state_confirmed=True,
)
def test_start_acceptance_requires_full_bootstrap_and_consumes_one_short_permit() -> None:
transport = SyntheticAcceptanceTransport()
permit = PhysicalAcceptancePermit(_checklist(ModelingAction.START))
executor = PhysicalAcceptanceDialogueExecutor(transport, permit)
orchestrator = ShadowApplicationBootstrapOrchestrator(
ApplicationControlAuthority(openapi_key=APPLICATION_KEY),
epoch_seconds=1_752_680_000,
timezone_name="Europe/Moscow",
)
binding = executor.run_bootstrap(orchestrator)
command = ShadowModelingCommand.from_command(
encode_modeling_start(
CommandHeaderIdentity(
device_id=binding.vendor_device_id,
openapi_key=APPLICATION_KEY,
),
project_name="SAFE_PROJECT",
record_mode=RecordMode.RECORD_AND_CALCULATE,
scan_mode=ScanMode.LCC,
mount_type=MountType.HANDHELD,
)
)
response = executor.execute_modeling(command)
assert response.action is ModelingAction.START
assert [len(batch) for batch in transport.batches] == [1, 2, 3, 1, 3, 1]
assert permit.snapshot()["consumed"] is True
assert executor.snapshot()["bootstrap_complete"] is True
assert executor.snapshot()["command_complete"] is True
response_evidence = executor.snapshot()["response_evidence"]
assert isinstance(response_evidence, tuple)
assert len(response_evidence) == 10
assert response_evidence[0]["operation_key"] == "bootstrap:1:DeviceInfoRequest"
assert response_evidence[-1]["operation_key"] == "modeling:start"
assert all("payload" not in item for item in response_evidence)
assert APPLICATION_KEY not in str(executor.snapshot())
assert VENDOR_DEVICE_ID not in str(executor.snapshot())
with pytest.raises(ApplicationAcceptanceError, match="already attempted"):
executor.execute_modeling(command)
def test_bootstrap_correlation_failure_records_only_redacted_response_evidence() -> None:
class CorruptDeviceConfigTransport(SyntheticAcceptanceTransport):
def exchange_batch_once(
self,
envelopes: Sequence[OneShotPublishEnvelope],
*,
required_response_topics: Collection[str],
) -> dict[str, bytes]:
responses = super().exchange_batch_once(
envelopes,
required_response_topics=required_response_topics,
)
if DEVICE_CONFIG_RESPONSE_TOPIC in responses:
responses[DEVICE_CONFIG_RESPONSE_TOPIC] = _generic_response(
f"{VENDOR_DEVICE_ID}:wrong-session"
)
return responses
executor = PhysicalAcceptanceDialogueExecutor(
CorruptDeviceConfigTransport(),
PhysicalAcceptancePermit(_checklist(ModelingAction.START)),
)
orchestrator = ShadowApplicationBootstrapOrchestrator(
ApplicationControlAuthority(openapi_key=APPLICATION_KEY),
epoch_seconds=1_752_680_000,
timezone_name="Europe/Moscow",
)
with pytest.raises(
ApplicationAcceptanceError,
match=r"ordinal 4 \(DeviceConfigRequest\).*session correlation",
):
executor.run_bootstrap(orchestrator)
snapshot = executor.snapshot()
failure = snapshot["correlation_failure"]
assert failure == {
"phase": "bootstrap",
"operation_key": "bootstrap:4:DeviceConfigRequest",
"response_topic": DEVICE_CONFIG_RESPONSE_TOPIC,
"reason": "application response session correlation failed",
}
evidence = snapshot["response_evidence"]
assert isinstance(evidence, tuple)
assert evidence[-1]["payload_sha256"] == hashlib.sha256(
_generic_response(f"{VENDOR_DEVICE_ID}:wrong-session")
).hexdigest()
assert evidence[-1]["payload_bytes"] > 0
assert all("payload" not in item for item in evidence)
assert APPLICATION_KEY not in str(snapshot)
assert VENDOR_DEVICE_ID not in str(snapshot)
def test_start_cannot_skip_bootstrap_and_stop_uses_a_separate_action_permit() -> None:
transport = SyntheticAcceptanceTransport()
start = PhysicalAcceptanceDialogueExecutor(
transport,
PhysicalAcceptancePermit(_checklist(ModelingAction.START)),
)
start_command = ShadowModelingCommand.from_command(
encode_modeling_start(
CommandHeaderIdentity(device_id=VENDOR_DEVICE_ID, openapi_key=APPLICATION_KEY),
project_name="SAFE_PROJECT",
record_mode=RecordMode.RECORD_AND_CALCULATE,
scan_mode=ScanMode.LCC,
mount_type=MountType.HANDHELD,
)
)
with pytest.raises(ApplicationAcceptanceError, match="requires.*bootstrap"):
start.execute_modeling(start_command)
stop = PhysicalAcceptanceDialogueExecutor(
transport,
PhysicalAcceptancePermit(_checklist(ModelingAction.STOP)),
)
stop_command = ShadowModelingCommand.from_command(
encode_modeling_stop(
CommandHeaderIdentity(device_id=VENDOR_DEVICE_ID, openapi_key=APPLICATION_KEY)
)
)
assert stop.execute_modeling(stop_command).action is ModelingAction.STOP
assert transport.batches[-1] == ("modeling:stop",)
def test_acceptance_permit_rejects_implicit_or_expired_authority() -> None:
with pytest.raises(ApplicationAcceptanceError, match="explicitly true"):
PhysicalAcceptanceChecklist(
action=ModelingAction.START,
operator_present=False, # type: ignore[arg-type]
owner_controlled_device=True,
lixelgo_closed=True,
battery_storage_confirmed=True,
expected_physical_state_confirmed=True,
)
now = [100.0]
permit = PhysicalAcceptancePermit(
_checklist(ModelingAction.STOP),
ttl_seconds=15.0,
monotonic=lambda: now[0],
)
now[0] = 115.0
with pytest.raises(ApplicationAcceptanceError, match="expired"):
permit.consume(ModelingAction.STOP)