Cognitive_Agent/tests/unit/test_vehicle_state_monitor.py

from __future__ import annotations

import threading
import time
import unittest
from collections import deque
from pathlib import Path
from types import SimpleNamespace

from cognitive_agent.application.seat_service import SeatDemoService
from cognitive_agent.application.vehicle_state_monitor import (
    OBSERVABLE_IDS,
    REQUIRED_UI_IDS,
    VehicleStateMonitor,
    VehicleStateSnapshot,
)
from cognitive_agent.infrastructure.can.gateway import (
    CanFrame,
    CanGatewayError,
    ZlgChannelConfig,
)

REPOSITORY_ROOT = Path(__file__).resolve().parents[2]


class _FakeBus:
    def __init__(self, frames: list[CanFrame]) -> None:
        self._frames = deque(frames)
        self._condition = threading.Condition()

    def push(self, frame: CanFrame) -> None:
        with self._condition:
            self._frames.append(frame)
            self._condition.notify_all()

    def receive(self, timeout_ms: int, *, include_fd: bool = False) -> CanFrame | None:
        if not include_fd:
            raise AssertionError("vehicle monitor must receive classic CAN and CAN FD")
        with self._condition:
            if not self._frames:
                self._condition.wait(max(0, timeout_ms) / 1000)
            return self._frames.popleft() if self._frames else None


class _FakeBridge:
    def __init__(self, bus: _FakeBus) -> None:
        self.bus = bus
        self.closed = False

    def channel_bus(self, channel: int) -> _FakeBus:
        if channel != 1:
            raise AssertionError("vehicle monitor must stay on CH1")
        return self.bus

    def close(self) -> None:
        self.closed = True


class _FakeSettings:
    channel = 1
    feedback_timeout_ms = 100

    def __init__(self, bus: _FakeBus) -> None:
        self.bus = bus
        self.bridge_instance = _FakeBridge(bus)
        self.monitor_only: bool | None = None

    def channel_config(self) -> ZlgChannelConfig:
        return ZlgChannelConfig(channel=1, termination_enabled=False)

    def bridge(
        self,
        channels: list[ZlgChannelConfig],
        *,
        monitor_only: bool = False,
    ) -> _FakeBridge:
        if len(channels) != 1:
            raise AssertionError("vehicle monitor must own exactly one channel")
        self.monitor_only = monitor_only
        if channels[0].receive_ids != tuple(sorted(OBSERVABLE_IDS)):
            raise AssertionError("vehicle monitor must filter to observable identifiers")
        return self.bridge_instance


def _frame(frame_id: int, data: bytes | None = None) -> CanFrame:
    return CanFrame(
        frame_id,
        data or bytes(8),
        channel=1,
        is_fd=True,
        bitrate_switch=True,
    )


def _baseline_frames() -> list[CanFrame]:
    gear_p = bytearray(8)
    gear_p[3] = 1
    return [
        _frame(0x342, bytes(gear_p)),
        _frame(0x4EB, bytes.fromhex("0700000000000000")),
        _frame(0x407),
        _frame(0x4FA),
        _frame(0x494),
    ]


class VehicleStateMonitorTests(unittest.TestCase):
    def test_persistent_monitor_returns_cached_state_without_reopening_adapter(self) -> None:
        bus = _FakeBus(_baseline_frames())
        settings = _FakeSettings(bus)
        monitor = VehicleStateMonitor(settings, reconnect_delay_s=0.01)  # type: ignore[arg-type]
        self.addCleanup(monitor.close)

        first = monitor.snapshot()
        started = time.perf_counter()
        second = monitor.snapshot()
        cached_latency_s = time.perf_counter() - started

        self.assertEqual(first.observed_ids, REQUIRED_UI_IDS)
        self.assertEqual(second.observed_ids, REQUIRED_UI_IDS)
        self.assertLess(cached_latency_s, 0.05)
        self.assertTrue(settings.monitor_only)
        self.assertEqual(monitor.public_info()["mode"], "persistent_monitor")

    def test_monitor_updates_sequence_for_each_new_feedback_frame(self) -> None:
        bus = _FakeBus(_baseline_frames())
        monitor = VehicleStateMonitor(
            _FakeSettings(bus),  # type: ignore[arg-type]
            reconnect_delay_s=0.01,
        )
        self.addCleanup(monitor.close)
        baseline = monitor.snapshot()

        gear_d = bytearray(8)
        gear_d[3] = 4
        bus.push(_frame(0x342, bytes(gear_d)))
        deadline = time.monotonic() + 0.5
        updated = baseline
        while updated.capture_sequence == baseline.capture_sequence:
            if time.monotonic() >= deadline:
                self.fail("monitor did not consume the new gear frame")
            time.sleep(0.005)
            updated = monitor.snapshot()

        gear = next(
            state["vehicle_safety"]["gear"]
            for state in updated.feedback_states
            if state["frame_id"] == "0x342"
        )
        self.assertEqual(gear, "D")

    def test_monitor_surfaces_adapter_open_failure(self) -> None:
        settings = _FakeSettings(_FakeBus([]))

        def fail_bridge(*_args: object, **_kwargs: object) -> _FakeBridge:
            raise RuntimeError("adapter open failed")

        settings.bridge = fail_bridge  # type: ignore[method-assign]
        monitor = VehicleStateMonitor(settings, reconnect_delay_s=0.01)  # type: ignore[arg-type]
        self.addCleanup(monitor.close)

        with self.assertRaisesRegex(CanGatewayError, "adapter open failed"):
            monitor.snapshot()

    def test_service_merges_a_cached_monitor_snapshot(self) -> None:
        service = SeatDemoService(REPOSITORY_ROOT)
        gear_state = {
            "source": "physical_can_ch1",
            "channel": 1,
            "frame_id": "0x342",
            "feedback_frame_id": "0x342",
            "valid": True,
            "captured_at": "2026-08-18T00:00:00Z",
            "vehicle_safety": {"gear": "P"},
            "verification": {
                "authoritative": True,
                "channel": 1,
                "simulated": False,
                "read_only": True,
                "frame_id": "0x342",
                "payload_size": 8,
            },
        }
        service.vehicle_can_settings = SimpleNamespace(
            state_configured=True,
            backend="zlgcan-linux",
        )
        service._vehicle_state_monitor = SimpleNamespace(
            snapshot=lambda: VehicleStateSnapshot(
                feedback_states=(gear_state,),
                observed_ids=REQUIRED_UI_IDS,
                captured_at="2026-08-18T00:00:00Z",
                capture_sequence=42,
            )
        )

        result = service._synchronize_vehicle_state_once(
            "conv-monitor-test",
            "denza_n7",
        )

        self.assertEqual(result["capture_sequence"], 42)
        self.assertEqual(result["domain_state"]["vehicle_status"]["gear"], "P")
        self.assertTrue(result["complete"])
        self.assertTrue(result["body_state_complete"])


if __name__ == "__main__":
    unittest.main()