from __future__ import annotations import os import tempfile 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): def setUp(self) -> None: self._temporary_directory = tempfile.TemporaryDirectory() self.addCleanup(self._temporary_directory.cleanup) self._linux_library = Path(self._temporary_directory.name) / "libusbcanfd.so" self._linux_library.touch() def _linux_environment(self) -> 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(self._linux_library), "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()