Skip to content

Commit ba289ee

Browse files
author
Pierre-Luc Tessier Gagne
committed
test(neovi): add unit tests for NeoViBus channel to netid conversion and message processing
1 parent 61e8f6f commit ba289ee

1 file changed

Lines changed: 165 additions & 0 deletions

File tree

‎test/test_neovi.py‎

Lines changed: 165 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,8 +4,15 @@
44

55
import pickle
66
import unittest
7+
from collections import deque
8+
from contextlib import ExitStack
9+
from threading import Event
10+
from types import SimpleNamespace
11+
from unittest.mock import patch
712

813
from can.interfaces.ics_neovi import ICSApiError
14+
from can.interfaces.ics_neovi.neovi_bus import NeoViBus
15+
from can.interfaces.ics_neovi import neovi_bus
916

1017

1118
class ICSApiErrorTest(unittest.TestCase):
@@ -22,5 +29,163 @@ def test_error_pickling(self):
2229
assert iae.__dict__ == un_pickled_iae.__dict__
2330

2431

32+
class NeoViBusBehaviorTest(unittest.TestCase):
33+
def test_channel_to_netid_accepts_integer_and_named_channel(self):
34+
fake_ics = SimpleNamespace(NETID_HSCAN=42)
35+
36+
with ExitStack() as stack:
37+
stack.enter_context(patch.object(neovi_bus, "ics", fake_ics))
38+
stack.enter_context(
39+
patch.object(neovi_bus, "ICS_NETID_LOOKUP", {"HSCAN": 42}, create=True)
40+
)
41+
42+
self.assertEqual(NeoViBus.channel_to_netid(7), 7)
43+
self.assertEqual(NeoViBus.channel_to_netid("8"), 8)
44+
self.assertEqual(NeoViBus.channel_to_netid("hscan"), 42)
45+
46+
def test_channel_to_netid_rejects_unknown_channel_name(self):
47+
fake_ics = SimpleNamespace(NETID_HSCAN=42)
48+
49+
with ExitStack() as stack:
50+
stack.enter_context(patch.object(neovi_bus, "ics", fake_ics))
51+
stack.enter_context(
52+
patch.object(neovi_bus, "ICS_NETID_LOOKUP", {"HSCAN": 42}, create=True)
53+
)
54+
55+
with self.assertRaises(ValueError):
56+
NeoViBus.channel_to_netid("unknown")
57+
58+
def test_ics_msg_to_message_converts_classic_frame(self):
59+
fake_ics = SimpleNamespace(
60+
SPY_PROTOCOL_CANFD=99,
61+
SPY_STATUS_XTD_FRAME=0x01,
62+
SPY_STATUS_REMOTE_FRAME=0x02,
63+
SPY_STATUS_TX_MSG=0x04,
64+
SPY_STATUS2_ERROR_FRAME=0x08,
65+
SPY_STATUS3_CANFD_ESI=0x10,
66+
SPY_STATUS3_CANFD_BRS=0x20,
67+
)
68+
bus = NeoViBus.__new__(NeoViBus)
69+
bus._use_system_timestamp = True
70+
bus._is_shutdown = True
71+
72+
ics_msg = SimpleNamespace(
73+
Protocol=0,
74+
StatusBitField=fake_ics.SPY_STATUS_XTD_FRAME,
75+
StatusBitField2=0,
76+
StatusBitField3=0,
77+
NumberBytesData=4,
78+
NetworkID=0x34,
79+
NetworkID2=0x12,
80+
ArbIDOrHeader=0x123,
81+
ExtraDataPtrEnabled=0,
82+
ExtraDataPtr=tuple(),
83+
Data=(1, 2, 3, 4, 9, 9, 9, 9),
84+
TimeSystem=12.5,
85+
)
86+
87+
with patch.object(neovi_bus, "ics", fake_ics):
88+
msg = bus._ics_msg_to_message(ics_msg)
89+
90+
self.assertEqual(msg.timestamp, 12.5)
91+
self.assertEqual(msg.arbitration_id, 0x123)
92+
self.assertTrue(msg.is_extended_id)
93+
self.assertFalse(msg.is_remote_frame)
94+
self.assertFalse(msg.is_error_frame)
95+
self.assertFalse(msg.is_fd)
96+
self.assertTrue(msg.is_rx)
97+
self.assertEqual(msg.channel, 0x1234)
98+
self.assertEqual(msg.dlc, 4)
99+
self.assertEqual(bytes(msg.data), b"\x01\x02\x03\x04")
100+
101+
def test_ics_msg_to_message_converts_fd_frame(self):
102+
fake_ics = SimpleNamespace(
103+
SPY_PROTOCOL_CANFD=99,
104+
SPY_STATUS_XTD_FRAME=0x01,
105+
SPY_STATUS_REMOTE_FRAME=0x02,
106+
SPY_STATUS_TX_MSG=0x04,
107+
SPY_STATUS2_ERROR_FRAME=0x08,
108+
SPY_STATUS3_CANFD_ESI=0x10,
109+
SPY_STATUS3_CANFD_BRS=0x20,
110+
)
111+
bus = NeoViBus.__new__(NeoViBus)
112+
bus._use_system_timestamp = True
113+
bus._is_shutdown = True
114+
115+
ics_msg = SimpleNamespace(
116+
Protocol=fake_ics.SPY_PROTOCOL_CANFD,
117+
StatusBitField=0,
118+
StatusBitField2=fake_ics.SPY_STATUS2_ERROR_FRAME,
119+
StatusBitField3=fake_ics.SPY_STATUS3_CANFD_BRS
120+
| fake_ics.SPY_STATUS3_CANFD_ESI,
121+
NumberBytesData=12,
122+
NetworkID=5,
123+
NetworkID2=0,
124+
ArbIDOrHeader=0x456,
125+
ExtraDataPtrEnabled=1,
126+
ExtraDataPtr=tuple(range(16)),
127+
Data=tuple(range(8)),
128+
TimeSystem=3.25,
129+
)
130+
131+
with patch.object(neovi_bus, "ics", fake_ics):
132+
msg = bus._ics_msg_to_message(ics_msg)
133+
134+
self.assertEqual(msg.timestamp, 3.25)
135+
self.assertEqual(msg.arbitration_id, 0x456)
136+
self.assertFalse(msg.is_extended_id)
137+
self.assertFalse(msg.is_remote_frame)
138+
self.assertTrue(msg.is_error_frame)
139+
self.assertTrue(msg.is_fd)
140+
self.assertTrue(msg.is_rx)
141+
self.assertTrue(msg.bitrate_switch)
142+
self.assertTrue(msg.error_state_indicator)
143+
self.assertEqual(msg.channel, 5)
144+
self.assertEqual(msg.dlc, 12)
145+
self.assertEqual(bytes(msg.data), bytes(range(12)))
146+
147+
def test_recv_internal_returns_none_when_no_message_available(self):
148+
bus = NeoViBus.__new__(NeoViBus)
149+
bus._is_shutdown = True
150+
bus.rx_buffer = deque()
151+
bus._process_msg_queue = lambda timeout=0.1: None
152+
153+
msg, already_filtered = bus._recv_internal(timeout=0)
154+
155+
self.assertIsNone(msg)
156+
self.assertFalse(already_filtered)
157+
158+
def test_process_msg_queue_sets_receipt_without_echoing_transmit(self):
159+
fake_ics = SimpleNamespace(
160+
SPY_STATUS_TX_MSG=0x01,
161+
SPY_STATUS_GLOBAL_ERR=0x02,
162+
get_messages=lambda dev, include_errors, timeout: ((tx_msg,), 0),
163+
)
164+
tx_msg = SimpleNamespace(
165+
NetworkID=1,
166+
NetworkID2=0,
167+
StatusBitField=fake_ics.SPY_STATUS_TX_MSG,
168+
ArbIDOrHeader=0x321,
169+
DescriptionID=17,
170+
)
171+
receipt_key = (tx_msg.ArbIDOrHeader, tx_msg.DescriptionID)
172+
receipt_event = Event()
173+
bus = NeoViBus.__new__(NeoViBus)
174+
bus._is_shutdown = False
175+
bus.dev = object()
176+
bus.channels = [1]
177+
bus._channel_set = {1}
178+
bus.rx_buffer = deque()
179+
bus.message_receipts = {receipt_key: receipt_event}
180+
bus._receive_own_messages = False
181+
182+
with patch.object(neovi_bus, "ics", fake_ics):
183+
bus._process_msg_queue(timeout=0)
184+
185+
self.assertTrue(receipt_event.is_set())
186+
self.assertEqual(len(bus.rx_buffer), 0)
187+
bus._is_shutdown = True
188+
189+
25190
if __name__ == "__main__":
26191
unittest.main()

0 commit comments

Comments
 (0)