vee
144 lines
5.0 KiB
Python
144 lines
5.0 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from cognitive_agent.application.native_can_feedback import (
|
|
VehicleSafetyPreflightError,
|
|
observe_vehicle_safety,
|
|
require_stationary_vehicle,
|
|
)
|
|
from cognitive_agent.application.seat_service import REPOSITORY_ROOT, SeatDemoService
|
|
from cognitive_agent.infrastructure.can.gateway import CanFrame
|
|
from cognitive_agent.infrastructure.can.hardware import HardwareSafetyError
|
|
|
|
|
|
class _ReceiveOnlyBus:
|
|
def __init__(self, frames: list[CanFrame]) -> None:
|
|
self.frames = list(frames)
|
|
self.send_called = False
|
|
|
|
def send(self, *args, **kwargs): # pragma: no cover - a safety tripwire
|
|
self.send_called = True
|
|
raise AssertionError("passive safety preflight must never transmit")
|
|
|
|
def receive(self, timeout_ms: int, *, include_fd: bool = False):
|
|
return self.frames.pop(0) if self.frames else None
|
|
|
|
def close(self) -> None:
|
|
return None
|
|
|
|
|
|
class _Bridge:
|
|
def __init__(self, bus: _ReceiveOnlyBus) -> None:
|
|
self.bus = bus
|
|
self.closed = False
|
|
|
|
def channel_bus(self, channel: int) -> _ReceiveOnlyBus:
|
|
if channel != 1:
|
|
raise AssertionError(f"unexpected channel {channel}")
|
|
return self.bus
|
|
|
|
def close(self) -> None:
|
|
self.closed = True
|
|
|
|
|
|
def _gear(raw: int, *, channel: int = 1) -> CanFrame:
|
|
data = bytearray(8)
|
|
data[3] = raw & 0x0F
|
|
return CanFrame(0x342, bytes(data), channel=channel)
|
|
|
|
|
|
def _speed(raw: int, *, channel: int = 1) -> CanFrame:
|
|
data = bytearray(8)
|
|
data[0] = 0x07
|
|
data[1] = raw & 0xFF
|
|
data[2] = (raw >> 8) & 0x0F
|
|
return CanFrame(0x4EB, bytes(data), channel=channel)
|
|
|
|
|
|
class VehicleSafetyPreflightTests(unittest.TestCase):
|
|
def test_fresh_ch1_p_and_zero_speed_pass_without_transmitting(self) -> None:
|
|
bus = _ReceiveOnlyBus([_gear(1), _speed(0)])
|
|
snapshot = observe_vehicle_safety(bus, timeout_ms=100)
|
|
require_stationary_vehicle(snapshot)
|
|
|
|
self.assertTrue(snapshot["stationary"])
|
|
self.assertEqual(snapshot["evidence"]["gear"]["frame_id"], "0x342")
|
|
self.assertEqual(snapshot["evidence"]["speed"]["frame_id"], "0x4EB")
|
|
self.assertFalse(bus.send_called)
|
|
|
|
def test_ch0_frames_cannot_satisfy_the_gate(self) -> None:
|
|
snapshot = observe_vehicle_safety(
|
|
_ReceiveOnlyBus([_gear(1, channel=0), _speed(0, channel=0)]),
|
|
timeout_ms=100,
|
|
)
|
|
|
|
self.assertFalse(snapshot["complete"])
|
|
self.assertEqual(snapshot["ignored_wrong_channel"], 2)
|
|
with self.assertRaisesRegex(
|
|
VehicleSafetyPreflightError, "0x342 gear evidence"
|
|
):
|
|
require_stationary_vehicle(snapshot)
|
|
|
|
def test_drive_or_nonzero_speed_fails_closed(self) -> None:
|
|
for frames, message in (
|
|
([_gear(4), _speed(0)], "not confirmed in P"),
|
|
([_gear(1), _speed(10)], "not confirmed at 0 km/h"),
|
|
):
|
|
with self.subTest(message=message):
|
|
snapshot = observe_vehicle_safety(
|
|
_ReceiveOnlyBus(frames), timeout_ms=100
|
|
)
|
|
with self.assertRaisesRegex(VehicleSafetyPreflightError, message):
|
|
require_stationary_vehicle(snapshot)
|
|
|
|
def test_policy_cannot_be_redirected_to_ch0(self) -> None:
|
|
with self.assertRaisesRegex(VehicleSafetyPreflightError, "only CH1"):
|
|
observe_vehicle_safety(
|
|
_ReceiveOnlyBus([_gear(1, channel=0), _speed(0, channel=0)]),
|
|
timeout_ms=100,
|
|
required_channel=0,
|
|
)
|
|
|
|
def test_service_refuses_before_any_command_when_vehicle_is_not_in_p(self) -> None:
|
|
service = SeatDemoService(REPOSITORY_ROOT)
|
|
action = {
|
|
"domain": "climate",
|
|
"capability": "climate.set_power",
|
|
"target": "vehicle",
|
|
"operation": "set",
|
|
"parameters": {"state": "on"},
|
|
}
|
|
config = service.native_can_runtime.build_function_config(
|
|
"climate", [action], "req-safety-preflight", "local_rules"
|
|
)
|
|
plan = service.native_can_runtime.compile(config)
|
|
bus = _ReceiveOnlyBus([_gear(4), _speed(0)])
|
|
bridge = _Bridge(bus)
|
|
|
|
with patch.dict(
|
|
"os.environ",
|
|
{"COGNITIVE_AGENT_SHARED_CAN_RUNTIME_ENABLED": "false"},
|
|
clear=False,
|
|
), patch.object(
|
|
type(service.vehicle_can_settings), "bridge", return_value=bridge
|
|
):
|
|
with self.assertRaisesRegex(HardwareSafetyError, "not confirmed in P"):
|
|
service._execute_orchestrated_reports(
|
|
{"native_climate": config},
|
|
{"native_climate": plan},
|
|
{"native_climate"},
|
|
{},
|
|
hardware_request=True,
|
|
simulated_hardware_request=False,
|
|
simulation_fault="normal",
|
|
)
|
|
|
|
self.assertFalse(bus.send_called)
|
|
self.assertTrue(bridge.closed)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|