NODEDC_MISSION_CORE/tests/test_xgrids_modeling_safety.py

from __future__ import annotations

import pytest

from k1link.device_plugins.xgrids_k1.protocol.application_bootstrap import (
    ApplicationControlAuthority,
    LiveDeviceControlBinding,
)
from k1link.device_plugins.xgrids_k1.protocol.modeling_control import (
    MODELING_STATE_BASE,
    ModelingAction,
    SessionState,
)
from k1link.device_plugins.xgrids_k1.protocol.modeling_safety import (
    DEVICE_STATUS_TOPIC,
    LiveModelingControlSafety,
    ModelingControlSafetyError,
)
from k1link.device_plugins.xgrids_k1.protocol.protobuf_wire import iter_fields
from k1link.device_plugins.xgrids_k1.viewer.messages import StreamMessage
from k1link.viewer.metrics import BridgeMetrics


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 _status_message(
    state: SessionState,
    *,
    device_id: str = "vendor-device-001",
    device_serial: str = "serial-001",
    project_id: str | None = None,
    source: str = "live_mqtt",
) -> StreamMessage:
    header = _text(4, device_id) + _text(5, "status-message-session")
    payload = _bytes(1, header) + _uint(2, MODELING_STATE_BASE + state) + _text(3, device_serial)
    if project_id is not None:
        payload += _text(4, project_id)
    return StreamMessage(
        sequence=1,
        topic=DEVICE_STATUS_TOPIC,
        payload=payload,
        received_at_epoch_ns=1,
        received_monotonic_ns=1,
        source=source,
    )


def _authority() -> ApplicationControlAuthority:
    return ApplicationControlAuthority(openapi_key="11111111-2222-3333-4444-555555555555")


def _binding(**changes: object) -> LiveDeviceControlBinding:
    values = {
        "vendor_device_id": "vendor-device-001",
        "device_serial": "serial-001",
        "software_version": "V3.0.2-build.1",
        "system_version": "V3.0.2",
        "device_model": "LixelKity K1",
        "device_type": "A4",
        "is_activated": True,
    }
    values.update(changes)
    return LiveDeviceControlBinding(**values)  # type: ignore[arg-type]


def test_live_status_identity_is_observed_without_accepting_replay() -> None:
    safety = LiveModelingControlSafety()
    metrics = BridgeMetrics()

    assert not safety.observe(
        _status_message(SessionState.READY, source="k1mqtt"),
        metrics,
    )
    assert safety.snapshot().status_reports_observed == 0

    assert safety.observe(_status_message(SessionState.READY), metrics)
    snapshot = safety.snapshot()
    assert snapshot.status_reports_observed == 1
    assert snapshot.vendor_identity_observed
    assert snapshot.device_serial_observed
    assert snapshot.ready_for_start_shadow
    assert not snapshot.ready_for_stop_shadow
    assert "vendor-device" not in repr(snapshot)


def test_start_shadow_plan_is_exact_bounded_and_never_executable() -> None:
    safety = LiveModelingControlSafety()
    safety.observe(_status_message(SessionState.READY), BridgeMetrics())

    plan = safety.plan_start(_authority(), _binding(), project_name="SAFE_PROJECT")
    fields = {field.number: field for field in iter_fields(plan.command.payload, max_fields=16)}

    assert plan.action is ModelingAction.START
    assert fields[2].value == ModelingAction.START
    assert fields[3].value == b"SAFE_PROJECT"
    assert fields[4].value == 2
    assert fields[5].value == 1
    assert 6 not in fields
    assert plan.qos == 2
    assert plan.retain is False
    assert plan.executable is False
    assert plan.blockers == ("vendor-writes-disabled", "publisher-not-installed")
    assert plan.retry_policy == "never-automatic"
    assert "11111111-2222-3333-4444-555555555555" not in repr(plan)
    assert "11111111-2222-3333-4444-555555555555" not in str(plan.as_dict())


def test_stop_shadow_plan_requires_same_live_device_and_active_project() -> None:
    safety = LiveModelingControlSafety()
    safety.observe(
        _status_message(SessionState.SCANNING, project_id="private-project-id"),
        BridgeMetrics(),
    )

    plan = safety.plan_stop(_authority(), _binding())
    fields = list(iter_fields(plan.command.payload, max_fields=8))
    assert plan.action is ModelingAction.STOP
    assert [field.number for field in fields] == [1, 2]

    with pytest.raises(ModelingControlSafetyError, match="does not match"):
        safety.plan_stop(_authority(), _binding(device_serial="another-serial"))


def test_identity_drift_and_decode_failure_fail_closed() -> None:
    safety = LiveModelingControlSafety()
    metrics = BridgeMetrics()
    safety.observe(_status_message(SessionState.READY), metrics)
    safety.observe(
        _status_message(SessionState.READY, device_id="different-vendor-device"),
        metrics,
    )
    assert safety.snapshot().identity_conflict
    with pytest.raises(ModelingControlSafetyError, match="conflicted"):
        safety.plan_start(_authority(), _binding(), project_name="SAFE_PROJECT")

    safety.reset()
    malformed = StreamMessage(
        sequence=2,
        topic=DEVICE_STATUS_TOPIC,
        payload=b"\x10",
        received_at_epoch_ns=2,
        received_monotonic_ns=2,
        source="live_mqtt",
    )
    assert safety.observe(malformed, metrics)
    assert safety.snapshot().decode_errors == 1
    with pytest.raises(ModelingControlSafetyError, match="conflicted"):
        safety.plan_start(_authority(), _binding(), project_name="SAFE_PROJECT")


def test_shadow_plan_requires_exact_safe_lifecycle_state() -> None:
    safety = LiveModelingControlSafety()
    safety.observe(_status_message(SessionState.SCAN_STARTING), BridgeMetrics())
    with pytest.raises(ModelingControlSafetyError, match="must be ready"):
        safety.plan_start(_authority(), _binding(), project_name="SAFE_PROJECT")
    with pytest.raises(ModelingControlSafetyError, match="must be scanning"):
        safety.plan_stop(_authority(), _binding())


def test_device_info_binding_rejects_unreviewed_or_inactive_profile() -> None:
    safety = LiveModelingControlSafety()
    safety.observe(_status_message(SessionState.READY), BridgeMetrics())

    with pytest.raises(ModelingControlSafetyError, match="does not attest"):
        safety.plan_start(
            _authority(),
            _binding(software_version="V3.0.3"),
            project_name="SAFE_PROJECT",
        )
    with pytest.raises(ModelingControlSafetyError, match="does not attest"):
        safety.plan_start(
            _authority(),
            _binding(is_activated=False),
            project_name="SAFE_PROJECT",
        )