Cognitive_Agent/tests/unit/test_wired_dual_channel_can.py

from __future__ import annotations

import os
import unittest
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch

from cognitive_agent.application.seat_service import REPOSITORY_ROOT, SeatDemoService


def _report(domain: str, request_id: str) -> dict[str, object]:
    return {
        "apiVersion": "cognitive-agent/v1alpha1",
        "kind": "ExecutionReport",
        "metadata": {
            "id": f"report-{domain}-{request_id}",
            "name": f"{domain} test report",
            "version": "1.0.0",
            "request_id": request_id,
            "labels": {"simulation": "false"},
        },
        "spec": {
            "mode": "hil",
            "status": "succeeded",
            "results": [
                {
                    "ref": f"{domain}-operation",
                    "domain": domain,
                    "status": "succeeded",
                    "message": "test",
                    "telemetry": {},
                }
            ],
        },
    }


class _ChannelBus:
    def __init__(self, channel: int) -> None:
        self.channel = channel

    def close(self) -> None:
        return None


class _Bridge:
    def __init__(self) -> None:
        self.buses = {0: _ChannelBus(0), 1: _ChannelBus(1)}
        self.closed = False

    def channel_bus(self, channel: int) -> _ChannelBus:
        return self.buses[channel]

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


class _NativeExecutor:
    def __init__(self, bus_factory, **_kwargs) -> None:
        self.bus_factory = bus_factory

    def execute(self, _runtime, _plan, config):
        if self.bus_factory().channel != 1:
            raise AssertionError("vehicle commands must use CAN1")
        return _report(config["spec"]["domain"], config["metadata"]["request_id"])


class _SeatExecutor:
    def __init__(self, bus_factory) -> None:
        self.bus_factory = bus_factory

    def execute(self, _plan, config, _state):
        if self.bus_factory().channel != 0:
            raise AssertionError("passenger-seat commands must use CAN0")
        report = _report("seat", config["metadata"]["request_id"])
        report["spec"]["physical_state"] = {
            "source": "physical_can",
            "channel": 0,
            "complete": False,
            "missing": ["longitudinal"],
            "position": {},
        }
        return report


class WiredDualChannelCanTests(unittest.TestCase):
    @staticmethod
    def _linux_environment() -> dict[str, str]:
        return {
            "COGNITIVE_AGENT_VEHICLE_CAN_ENABLED": "true",
            "COGNITIVE_AGENT_VEHICLE_CAN_BACKEND": "zlgcan-linux",
            "COGNITIVE_AGENT_WIRED_PASSENGER_CAN_ENABLED": "true",
            "COGNITIVE_AGENT_SEAT_CAN_CHANNEL": "0",
            "COGNITIVE_AGENT_VEHICLE_CAN_CHANNEL": "1",
            "COGNITIVE_AGENT_HARDWARE_APPROVAL_TOKEN_REQUIRED": "false",
            "COGNITIVE_AGENT_ZLGCAN_PYTHON": os.sys.executable,
            "COGNITIVE_AGENT_ZLGCAN_LINUX_LIBRARY": str(
                REPOSITORY_ROOT.parent / "project-code" / "canfd" / "libusbcanfd.so"
            ),
            "COGNITIVE_AGENT_SHARED_CAN_RUNTIME_ENABLED": "false",
        }

    def test_runtime_reports_manual_wired_dual_channel_policy(self) -> None:
        with patch.dict(os.environ, self._linux_environment(), clear=True):
            service = SeatDemoService(REPOSITORY_ROOT)
            runtime = service.runtime_info()

        wired = next(
            item
            for item in runtime["vehicle_transports"]["options"]
            if item["id"] == "wired_can"
        )
        self.assertEqual(
            wired["target_policy"],
            "dual_channel_wired_can0_passenger_can1_vehicle",
        )
        self.assertEqual(wired["channels"], {"seat": 0, "vehicle": 1})
        self.assertEqual(
            runtime["multi_channel_can"]["active_policy"],
            "dual_channel_wired_can0_passenger_can1_vehicle",
        )

    def test_linked_wired_execution_routes_seat_to_can0_and_vehicle_to_can1(self) -> None:
        with patch.dict(os.environ, self._linux_environment(), clear=True):
            service = SeatDemoService(Path(REPOSITORY_ROOT))
        bridge = _Bridge()
        request_id = "req-wired-dual-channel"
        seat_config = {
            "metadata": {"request_id": request_id},
            "spec": {"domain": "seat", "operations": []},
        }
        native_config = {
            "metadata": {"request_id": request_id},
            "spec": {"domain": "climate", "operations": []},
        }
        configs = {"seat": seat_config, "native_climate": native_config}
        plans = {
            "seat": {"spec": {"commands": []}},
            "native_climate": {"spec": {"commands": []}},
        }
        captured_channel_configs = []

        def create_bridge(channel_configs):
            captured_channel_configs.extend(channel_configs)
            return bridge

        def create_seat_executor(bus_factory, *, remote_transport=False):
            self.assertFalse(remote_transport)
            return _SeatExecutor(bus_factory)

        with (
            patch.dict(
                os.environ,
                {"COGNITIVE_AGENT_SHARED_CAN_RUNTIME_ENABLED": "false"},
                clear=False,
            ),
            patch.object(
                type(service.vehicle_can_settings),
                "bridge",
                side_effect=create_bridge,
            ),
            patch.object(
                service,
                "_new_hardware_executor",
                side_effect=create_seat_executor,
            ),
            patch(
                "cognitive_agent.application.seat_service.NativeCanExecutor",
                _NativeExecutor,
            ),
            patch(
                "cognitive_agent.application.seat_service.observe_vehicle_safety",
                return_value={"stationary": True},
            ),
            patch(
                "cognitive_agent.application.seat_service.require_stationary_vehicle"
            ),
        ):
            reports = service._execute_orchestrated_reports(
                configs,
                plans,
                {"native_climate"},
                {"devices": {}},
                hardware_request=True,
                simulated_hardware_request=False,
                simulation_fault="normal",
                control_transport="wired_can",
            )

        self.assertEqual(set(reports), {"seat", "native_climate"})
        self.assertEqual(
            [(item.channel, item.frame_format) for item in captured_channel_configs],
            [(0, "classic_can"), (1, "can_fd")],
        )
        self.assertTrue(bridge.closed)

    def test_wired_passenger_status_uses_monitor_only_can0_bridge(self) -> None:
        with patch.dict(os.environ, self._linux_environment(), clear=True):
            service = SeatDemoService(Path(REPOSITORY_ROOT))
        bridge = _Bridge()
        executor = SimpleNamespace(
            read_physical_state=lambda: {
                "position": {},
                "complete": False,
                "missing": ["longitudinal"],
                "captured_at": "2026-08-25T00:00:00Z",
            }
        )

        with (
            patch.object(
                type(service.vehicle_can_settings),
                "bridge",
                return_value=bridge,
            ) as create_bridge,
            patch.object(
                service,
                "_new_hardware_executor",
                return_value=executor,
            ) as create_executor,
        ):
            result = service._synchronize_hardware_state_once(
                "conv-wired-monitor-only",
                control_transport="wired_can",
            )

        channel_configs = create_bridge.call_args.args[0]
        self.assertTrue(create_bridge.call_args.kwargs["monitor_only"])
        self.assertEqual(
            [(item.channel, item.frame_format) for item in channel_configs],
            [(0, "classic_can")],
        )
        bus_factory = create_executor.call_args.args[0]
        self.assertEqual(bus_factory().channel, 0)
        self.assertEqual(result["source"], "physical_can")
        self.assertTrue(bridge.closed)


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