cmvr_edge_ai/tests/unit/test_safety_gate.py

from __future__ import annotations

import asyncio
from collections.abc import Mapping
from time import time_ns
from typing import Any

import pytest

from cmvr_edge_ai.contracts import ApprovedRobotCommand, RobotCommand
from cmvr_edge_ai.core import Envelope
from cmvr_edge_ai.plugins.builtin import CommandSafetyGate


def _envelope(
    action: str,
    *,
    valid_until_ns: int,
    sequence: int = 1,
    arguments: Mapping[str, Any] | None = None,
) -> Envelope[RobotCommand]:
    return Envelope(
        RobotCommand(
            device_id="src1100",
            action=action,
            valid_until_ns=valid_until_ns,
            sequence=sequence,
            arguments={} if arguments is None else arguments,
        ),
        schema_name="RobotCommand",
    )


def test_command_safety_gate_is_fail_closed() -> None:
    gate = CommandSafetyGate(
        "gate",
        {
            "allowed_actions": ["emergency_stop"],
            "allowed_devices": ["src1100"],
        },
    )
    future = time_ns() + 5_000_000_000

    command_envelope = _envelope("emergency_stop", valid_until_ns=future)
    accepted = asyncio.run(gate.process(command_envelope))
    expired = asyncio.run(
        gate.process(_envelope("emergency_stop", valid_until_ns=time_ns() - 1))
    )
    unauthorized = asyncio.run(
        gate.process(_envelope("set_velocity", valid_until_ns=future))
    )

    assert accepted is not None
    assert accepted.port == "commands"
    assert accepted.envelope.schema == "ApprovedRobotCommand/v1"
    assert isinstance(accepted.envelope.payload, ApprovedRobotCommand)
    assert accepted.envelope.payload.command is command_envelope.payload
    assert accepted.envelope.payload.approved_by == "gate"
    assert accepted.envelope.trace_id == command_envelope.trace_id
    assert expired is None
    assert unauthorized is None
    assert asyncio.run(gate.health()).detail == "rejected_commands=2"


def test_velocity_gate_requires_limits_and_rejects_replays() -> None:
    with pytest.raises(ValueError, match="requires argument_limits"):
        CommandSafetyGate("gate", {"allowed_actions": ["set_velocity"]})

    gate = CommandSafetyGate(
        "gate",
        {
            "allowed_actions": ["set_velocity"],
            "allowed_devices": ["src1100"],
            "argument_limits": {
                "set_velocity": {
                    "vx": {"min": -1.0, "max": 1.0},
                    "vy": {"min": -1.0, "max": 1.0},
                    "wz": {"min": -2.0, "max": 2.0},
                }
            },
        },
    )
    future = time_ns() + 5_000_000_000
    accepted = _envelope(
        "set_velocity",
        valid_until_ns=future,
        sequence=10,
        arguments={"vx": 0.5, "vy": 0.0, "wz": 0.25},
    )

    approved = asyncio.run(gate.process(accepted))
    assert approved is not None
    assert isinstance(approved.envelope.payload, ApprovedRobotCommand)
    assert asyncio.run(gate.process(accepted)) is None
    outside_limits = asyncio.run(
        gate.process(
            _envelope(
                "set_velocity",
                valid_until_ns=future,
                sequence=11,
                arguments={"vx": 1.5, "vy": 0.0, "wz": 0.0},
            )
        )
    )
    assert outside_limits is None
    assert asyncio.run(gate.health()).detail == "rejected_commands=2"