NODEDC_MISSION_CORE/tests/test_xgrids_application_bootstrap.py

from __future__ import annotations

import pytest

from k1link.device_plugins.xgrids_k1.protocol.application_bootstrap import (
    ApplicationBootstrapError,
    ApplicationControlAuthority,
    LiveDeviceControlBinding,
    build_shadow_bootstrap,
    decode_and_bind_device_info_response,
)
from k1link.device_plugins.xgrids_k1.protocol.modeling_control import OPENAPI_SUCCESS
from k1link.device_plugins.xgrids_k1.protocol.protobuf_wire import iter_fields

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 _authority() -> ApplicationControlAuthority:
    return ApplicationControlAuthority(openapi_key=APPLICATION_KEY)


def _binding(**changes: object) -> LiveDeviceControlBinding:
    values = {
        "vendor_device_id": VENDOR_DEVICE_ID,
        "device_serial": "K1SERIAL01",
        "software_version": "V3.0.2-20260101-release",
        "system_version": "V3.0.2",
        "device_model": "LixelKity K1",
        "device_type": "K1",
        "is_activated": True,
    }
    values.update(changes)
    return LiveDeviceControlBinding(**values)  # type: ignore[arg-type]


def _device_info_response(*, session_id: str = ":DeviceInfoRequest") -> bytes:
    header = _text(4, VENDOR_DEVICE_ID) + _text(5, session_id) + _text(6, APPLICATION_KEY)
    base_info = b"".join(
        (
            _text(1, "2026-01-01T00:00:00"),
            _text(2, "V3.0.2-20260101-release"),
            _text(3, "V3.0.2"),
            _text(4, "V1.2.3"),
            _text(5, "V1.0"),
            _text(6, "LixelKity K1"),
            _text(7, "K1SERIAL01"),
            _text(8, "K1"),
        )
    )
    working_status = _uint(1, 1) + _uint(2, 0)
    device_info = _bytes(2, base_info) + _bytes(7, working_status)
    error = _uint(1, OPENAPI_SUCCESS) + _text(2, "success")
    return _bytes(1, header) + _bytes(2, device_info) + _bytes(15, error)


def test_exact_clean_cycle_pre_start_order_and_wire_lengths() -> None:
    plan = build_shadow_bootstrap(
        _authority(),
        _binding(),
        epoch_seconds=1_752_680_000,
        timezone_name="Europe/Moscow",
    )

    assert [request.message_type for request in plan.requests] == [
        "DeviceInfoRequest",
        "ModelingStatusRequest",
        "GetRtkAdvanceRequest",
        "DeviceConfigRequest",
        "DeviceInfoRequest",
        "GetRtkAdvanceRequest",
        "GetNtripProfileRequest",
        "GetCloudServerConfigRequest",
        "GetRtkAdvanceRequest",
        "DeviceInfoRequest",
    ]
    assert [request.payload_bytes for request in plan.requests] == [
        60,
        64,
        63,
        195,
        135,
        138,
        140,
        145,
        138,
        135,
    ]
    assert sum(request.mutates_device for request in plan.requests) == 1
    assert plan.requests[3].mutates_device
    assert plan.executable is False
    assert plan.blockers == ("vendor-writes-disabled", "publisher-not-installed")
    assert plan.retry_policy == "never-automatic"
    assert APPLICATION_KEY not in repr(plan)
    assert APPLICATION_KEY not in str(plan.as_dict())


def test_bootstrap_headers_switch_from_unbound_to_live_device_identity() -> None:
    plan = build_shadow_bootstrap(
        _authority(),
        _binding(),
        epoch_seconds=1_752_680_000,
        timezone_name="Europe/Moscow",
    )

    first_header = next(iter_fields(plan.requests[0].payload)).value
    assert isinstance(first_header, bytes)
    first_fields = {field.number: field.value for field in iter_fields(first_header)}
    assert 4 not in first_fields
    assert first_fields[5] == b":DeviceInfoRequest"

    config_header = next(iter_fields(plan.requests[3].payload)).value
    assert isinstance(config_header, bytes)
    config_fields = {field.number: field.value for field in iter_fields(config_header)}
    assert config_fields[4] == VENDOR_DEVICE_ID.encode()
    assert (
        config_fields[5]
        == (f"{VENDOR_DEVICE_ID}:DeviceConfigRequest:Publish_Proto_DeviceConfig_SetTime").encode()
    )


def test_device_info_response_produces_live_binding_without_a_saved_device_profile() -> None:
    response = decode_and_bind_device_info_response(_device_info_response(), _authority())

    assert response.result_code == OPENAPI_SUCCESS
    assert response.binding.ready_for_reviewed_profile
    assert response.binding.matches_reviewed_firmware
    assert response.binding.is_activated
    assert VENDOR_DEVICE_ID not in repr(response)
    assert APPLICATION_KEY not in repr(response)


def test_device_info_response_correlation_and_profile_attestation_fail_closed() -> None:
    with pytest.raises(ApplicationBootstrapError, match="session correlation"):
        decode_and_bind_device_info_response(
            _device_info_response(session_id=":AnotherRequest"),
            _authority(),
        )

    with pytest.raises(ApplicationBootstrapError, match="does not attest"):
        build_shadow_bootstrap(
            _authority(),
            _binding(system_version="V3.0.3"),
            epoch_seconds=1_752_680_000,
            timezone_name="Europe/Moscow",
        )