minghanchen
144 lines
5.2 KiB
Python
144 lines
5.2 KiB
Python
from __future__ import annotations
|
|
|
|
import io
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import Mock, patch
|
|
|
|
from cognitive_agent.infrastructure.can.gateway import (
|
|
ZlgCanBridgeProcess,
|
|
ZlgChannelConfig,
|
|
)
|
|
from cognitive_agent.infrastructure.can.hardware import VehicleCanSettings
|
|
from cognitive_agent.infrastructure.can.zlgcan_linux_bridge import (
|
|
EXPECTED_ABI_SIZES,
|
|
Driver,
|
|
_acceptance_filters,
|
|
_verified_init,
|
|
abi_sizes,
|
|
)
|
|
|
|
REPOSITORY_ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
class LinuxZlgBridgeTests(unittest.TestCase):
|
|
def test_ctypes_layout_matches_vendor_packed_abi(self) -> None:
|
|
self.assertEqual(abi_sizes(), EXPECTED_ABI_SIZES)
|
|
|
|
def test_verified_60mhz_timing_matches_can_dev_baseline(self) -> None:
|
|
init = _verified_init(
|
|
{
|
|
"arbitration_bitrate": 500_000,
|
|
"data_bitrate": 2_000_000,
|
|
"can_fd_iso": True,
|
|
}
|
|
)
|
|
self.assertEqual(init.clk, 60_000_000)
|
|
self.assertEqual(init.mode, 0)
|
|
self.assertEqual(
|
|
(init.aset.tseg1, init.aset.tseg2, init.aset.sjw, init.aset.brp),
|
|
(14, 3, 2, 5),
|
|
)
|
|
self.assertEqual(
|
|
(init.dset.tseg1, init.dset.tseg2, init.dset.sjw, init.dset.brp),
|
|
(10, 2, 2, 1),
|
|
)
|
|
with self.assertRaisesRegex(ValueError, "locked"):
|
|
_verified_init(
|
|
{"arbitration_bitrate": 1_000_000, "data_bitrate": 4_000_000}
|
|
)
|
|
|
|
def test_passive_state_filter_uses_exact_standard_identifiers(self) -> None:
|
|
frame_ids = frozenset({0x342, 0x407, 0x4EB})
|
|
filters = _acceptance_filters({"receive_ids": frame_ids})
|
|
|
|
self.assertEqual(filters.size, 3 * 12)
|
|
self.assertEqual(
|
|
[
|
|
(
|
|
filters.table[index].type,
|
|
filters.table[index].sid,
|
|
filters.table[index].eid,
|
|
)
|
|
for index in range(3)
|
|
],
|
|
[(0, 0x342, 0x342), (0, 0x407, 0x407), (0, 0x4EB, 0x4EB)],
|
|
)
|
|
|
|
def test_channel_config_serializes_receive_filter_only_when_requested(self) -> None:
|
|
filtered = ZlgChannelConfig(channel=1, receive_ids=(0x342, 0x4EB))
|
|
unfiltered = ZlgChannelConfig(channel=1)
|
|
|
|
self.assertEqual(filtered.public()["receive_ids"], [0x342, 0x4EB])
|
|
self.assertNotIn("receive_ids", unfiltered.public())
|
|
|
|
def test_monitor_only_rejects_send_before_touching_sdk(self) -> None:
|
|
driver = object.__new__(Driver)
|
|
driver.monitor_only = True
|
|
driver.configs = {1: {"channel": 1, "frame_format": "can_fd"}}
|
|
driver.sdk = Mock()
|
|
|
|
with self.assertRaisesRegex(PermissionError, "monitor-only"):
|
|
driver.send(
|
|
1,
|
|
0x123,
|
|
bytes.fromhex("0001020304050607"),
|
|
is_fd=True,
|
|
bitrate_switch=True,
|
|
)
|
|
|
|
driver.sdk.VCI_TransmitFD.assert_not_called()
|
|
driver.sdk.VCI_Transmit.assert_not_called()
|
|
|
|
def test_linux_state_can_be_ready_while_transmit_stays_locked(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
library = Path(directory) / "libusbcanfd.so"
|
|
library.write_bytes(b"test")
|
|
environment = {
|
|
"COGNITIVE_AGENT_VEHICLE_CAN_ENABLED": "true",
|
|
"COGNITIVE_AGENT_VEHICLE_CAN_BACKEND": "zlgcan-linux",
|
|
"COGNITIVE_AGENT_VEHICLE_CAN_CHANNEL": "1",
|
|
"COGNITIVE_AGENT_VEHICLE_CAN_REQUIRED_CHANNEL": "1",
|
|
"COGNITIVE_AGENT_ZLGCAN_PYTHON": sys.executable,
|
|
"COGNITIVE_AGENT_ZLGCAN_LINUX_LIBRARY": str(library),
|
|
}
|
|
with patch.dict(os.environ, environment, clear=True):
|
|
settings = VehicleCanSettings.from_env(REPOSITORY_ROOT)
|
|
self.assertEqual(settings.device_type, 33)
|
|
self.assertTrue(settings.state_configured)
|
|
self.assertFalse(settings.configured)
|
|
self.assertTrue(settings.bridge_script.name.endswith("linux_bridge.py"))
|
|
self.assertTrue(settings.public_info()["state_configured"])
|
|
|
|
def test_process_passes_monitor_and_trace_flags_to_bridge(self) -> None:
|
|
process = Mock()
|
|
process.stdout = io.StringIO('{"ok":true,"event":"ready"}\n')
|
|
process.stderr = io.StringIO()
|
|
process.stdin = io.StringIO()
|
|
process.poll.return_value = 0
|
|
with patch(
|
|
"cognitive_agent.infrastructure.can.gateway.subprocess.Popen",
|
|
return_value=process,
|
|
) as popen:
|
|
bridge = ZlgCanBridgeProcess(
|
|
python_executable=Path(sys.executable),
|
|
bridge_script=Path("bridge.py"),
|
|
dll_path=Path("libusbcanfd.so"),
|
|
device_type=33,
|
|
device_index=0,
|
|
channels=[ZlgChannelConfig(channel=1, termination_enabled=False)],
|
|
monitor_only=True,
|
|
trace_all=True,
|
|
)
|
|
command = popen.call_args.args[0]
|
|
self.assertIn("--monitor-only", command)
|
|
self.assertIn("--trace-all", command)
|
|
bridge.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|