95 lines
2.8 KiB
Python
95 lines
2.8 KiB
Python
import can
|
|
import pytest
|
|
|
|
from lingkong_motor_service.config import Settings
|
|
from lingkong_motor_service.protocol import build_open_loop_frame
|
|
from lingkong_motor_service.transport import (
|
|
CanTransportError,
|
|
DisabledCanTransport,
|
|
PythonCanTransport,
|
|
)
|
|
|
|
|
|
class StubBus:
|
|
def __init__(self) -> None:
|
|
self.messages: list[can.Message] = []
|
|
self.timeouts: list[float | None] = []
|
|
self.shutdown_count = 0
|
|
|
|
def send(self, message: can.Message, timeout: float | None = None) -> None:
|
|
self.messages.append(message)
|
|
self.timeouts.append(timeout)
|
|
|
|
def shutdown(self) -> None:
|
|
self.shutdown_count += 1
|
|
|
|
|
|
def test_canalystii_configuration_and_standard_frame(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
captured: dict[str, object] = {}
|
|
bus = StubBus()
|
|
|
|
def open_bus(**kwargs: object) -> StubBus:
|
|
captured.update(kwargs)
|
|
return bus
|
|
|
|
monkeypatch.setattr(can, "Bus", open_bus)
|
|
transport = PythonCanTransport(Settings(hardware_enabled=True, can_channel=1, can_device=2))
|
|
|
|
try:
|
|
transport.send(build_open_loop_frame({1: 400, 2: -400}))
|
|
|
|
assert captured == {
|
|
"interface": "canalystii",
|
|
"channel": 1,
|
|
"device": 2,
|
|
"bitrate": 1_000_000,
|
|
"rx_queue_size": 256,
|
|
"receive_own_messages": False,
|
|
}
|
|
assert bus.timeouts == [None]
|
|
assert len(bus.messages) == 1
|
|
assert bus.messages[0].arbitration_id == 0x280
|
|
assert bus.messages[0].is_extended_id is False
|
|
assert bytes(bus.messages[0].data) == bytes.fromhex("90 01 70 FE 00 00 00 00")
|
|
finally:
|
|
transport.shutdown()
|
|
|
|
|
|
def test_disabled_transport_fails_closed() -> None:
|
|
with pytest.raises(CanTransportError, match="hardware access is disabled"):
|
|
DisabledCanTransport().send(build_open_loop_frame({1: 0}))
|
|
|
|
|
|
def test_canalystii_can_wait_for_the_hardware_tx_queue(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
bus = StubBus()
|
|
monkeypatch.setattr(can, "Bus", lambda **kwargs: bus)
|
|
transport = PythonCanTransport(Settings(hardware_enabled=True))
|
|
|
|
try:
|
|
transport.send(
|
|
build_open_loop_frame({}),
|
|
tx_queue_timeout_s=0.1,
|
|
)
|
|
|
|
assert bus.timeouts == [0.1]
|
|
finally:
|
|
transport.shutdown()
|
|
|
|
|
|
def test_shutdown_is_idempotent_and_send_after_shutdown_fails(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
bus = StubBus()
|
|
monkeypatch.setattr(can, "Bus", lambda **kwargs: bus)
|
|
transport = PythonCanTransport(Settings(hardware_enabled=True))
|
|
|
|
transport.shutdown()
|
|
transport.shutdown()
|
|
|
|
assert transport.is_open is False
|
|
assert bus.shutdown_count == 1
|
|
with pytest.raises(CanTransportError, match="CAN transport is closed"):
|
|
transport.send(build_open_loop_frame({1: 0}))
|