86 lines
2.8 KiB
Python
86 lines
2.8 KiB
Python
import pytest
|
|
|
|
from lingkong_motor_service.protocol import (
|
|
CONTROL_FRAME_ID,
|
|
MIXED_COMMAND_FRAME_ID,
|
|
MOTOR_REPLY_BASE_ID,
|
|
MS_OPEN_LOOP_MAX,
|
|
CanFrame,
|
|
MotorCommand,
|
|
MotorStatus2,
|
|
ProtocolError,
|
|
build_mixed_command_frame,
|
|
build_open_loop_frame,
|
|
build_status_2_request,
|
|
parse_status_2_reply,
|
|
)
|
|
|
|
|
|
def test_pdf_golden_vector_positive_100_motor_1_negative_100_motor_3() -> None:
|
|
frame = build_open_loop_frame({1: 100, 3: -100})
|
|
|
|
assert frame.arbitration_id == CONTROL_FRAME_ID
|
|
assert frame.is_extended_id is False
|
|
assert frame.data == bytes.fromhex("64 00 00 00 9C FF 00 00")
|
|
|
|
|
|
def test_requested_open_and_close_vectors_for_motors_1_and_2() -> None:
|
|
opening = build_open_loop_frame({1: -400, 2: 400})
|
|
closing = build_open_loop_frame({1: 400, 2: -400})
|
|
|
|
assert opening.data == bytes.fromhex("70 FE 90 01 00 00 00 00")
|
|
assert closing.data == bytes.fromhex("90 01 70 FE 00 00 00 00")
|
|
|
|
|
|
@pytest.mark.parametrize("value", [-851, 851])
|
|
def test_rejects_values_outside_ms_open_loop_range(value: int) -> None:
|
|
with pytest.raises(ProtocolError, match="MS open_loop_raw"):
|
|
build_open_loop_frame({1: value})
|
|
|
|
|
|
@pytest.mark.parametrize("value", [-MS_OPEN_LOOP_MAX, MS_OPEN_LOOP_MAX])
|
|
def test_accepts_ms_open_loop_range_boundaries(value: int) -> None:
|
|
frame = build_open_loop_frame({2: value})
|
|
assert len(frame.data) == 8
|
|
|
|
|
|
@pytest.mark.parametrize("motor_id", [0, 5, True])
|
|
def test_rejects_invalid_broadcast_motor_id(motor_id: int) -> None:
|
|
with pytest.raises(ProtocolError):
|
|
build_open_loop_frame({motor_id: 1})
|
|
|
|
|
|
def test_mixed_enable_stop_and_disable_vectors() -> None:
|
|
enable = build_mixed_command_frame({1: MotorCommand.ENABLE})
|
|
stop = build_mixed_command_frame({4: MotorCommand.STOP})
|
|
disable = build_mixed_command_frame({2: MotorCommand.DISABLE})
|
|
|
|
assert enable.arbitration_id == MIXED_COMMAND_FRAME_ID
|
|
assert enable.data == bytes.fromhex("88 00 00 00 00 00 00 00")
|
|
assert stop.data == bytes.fromhex("00 00 00 00 00 00 81 00")
|
|
assert disable.data == bytes.fromhex("00 00 80 00 00 00 00 00")
|
|
|
|
|
|
def test_status_2_request_and_reply_temperature() -> None:
|
|
request = build_status_2_request((1, 2))
|
|
reply = CanFrame(
|
|
arbitration_id=MOTOR_REPLY_BASE_ID + 2,
|
|
data=bytes.fromhex("9C 50 64 00 0A 00 34 12"),
|
|
)
|
|
|
|
assert request.arbitration_id == MIXED_COMMAND_FRAME_ID
|
|
assert request.data == bytes.fromhex("9C 00 9C 00 00 00 00 00")
|
|
assert parse_status_2_reply(reply) == MotorStatus2(
|
|
motor_id=2,
|
|
temperature_c=80,
|
|
power_raw=100,
|
|
speed_dps=10,
|
|
encoder_raw=0x1234,
|
|
)
|
|
|
|
|
|
def test_status_2_parser_ignores_unrelated_frame() -> None:
|
|
frame = CanFrame(arbitration_id=CONTROL_FRAME_ID, data=bytes(8))
|
|
|
|
assert parse_status_2_reply(frame) is None
|