668170457e
Migrates all 43 sim modules and 23 test files (~7.8k LOC, 686 tests) from .worktrees/sim-framework/ into pm3py/sim/. Import fixes: - 4 files: ..protocol/..transport → ..core.protocol/..core.transport - trace_fmt.py: pm3py.hf_15 → pm3py.trace.ndef - test_sim_pm3medium.py: flat imports → core.* - test_sim_trace_fmt.py: flat imports → core.* - Added sim Cmd entries to core/protocol.py (EML_SETMEM, SIM_TABLE_*) 751 tests passing (686 sim + 65 core). Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
252 lines
7.9 KiB
Python
252 lines
7.9 KiB
Python
import pytest
|
|
from pm3py.sim.mcu_protocol import (
|
|
cobs_encode, cobs_decode,
|
|
MsgType, build_frame, parse_frame,
|
|
StreamDeframer,
|
|
)
|
|
|
|
|
|
class TestCobsCodec:
|
|
def test_encode_empty(self):
|
|
assert cobs_encode(b"") == b"\x01"
|
|
|
|
def test_encode_no_zeros(self):
|
|
assert cobs_encode(b"\x01\x02\x03") == b"\x04\x01\x02\x03"
|
|
|
|
def test_encode_single_zero(self):
|
|
assert cobs_encode(b"\x00") == b"\x01\x01"
|
|
|
|
def test_encode_zeros_in_data(self):
|
|
assert cobs_encode(b"\x01\x00\x02") == b"\x02\x01\x02\x02"
|
|
|
|
def test_decode_roundtrip(self):
|
|
for data in [b"", b"\x00", b"\x01\x02\x03", b"\x00\x00\x00",
|
|
b"hello world", bytes(range(256))]:
|
|
assert cobs_decode(cobs_encode(data)) == data
|
|
|
|
def test_decode_empty_raises(self):
|
|
with pytest.raises(ValueError):
|
|
cobs_decode(b"")
|
|
|
|
def test_decode_invalid_raises(self):
|
|
with pytest.raises(ValueError):
|
|
cobs_decode(b"\xFF")
|
|
|
|
|
|
class TestMessageFraming:
|
|
def test_build_frame_no_payload(self):
|
|
frame = build_frame(MsgType.RESET)
|
|
assert frame[-1] == 0x00
|
|
decoded = cobs_decode(frame[:-1])
|
|
assert decoded == bytes([0xF0])
|
|
|
|
def test_build_frame_with_payload(self):
|
|
frame = build_frame(MsgType.SET_ED, bytes([0x01]))
|
|
decoded = cobs_decode(frame[:-1])
|
|
assert decoded == bytes([0x20, 0x01])
|
|
|
|
def test_parse_frame_i2c_write(self):
|
|
raw = bytes([MsgType.I2C_WRITE, 0x20, 0x00, 0xDE, 0xAD])
|
|
msg_type, payload = parse_frame(raw)
|
|
assert msg_type == MsgType.I2C_WRITE
|
|
assert payload == bytes([0x20, 0x00, 0xDE, 0xAD])
|
|
|
|
def test_parse_frame_mcu_ready(self):
|
|
raw = bytes([MsgType.MCU_READY, 0x01, 0x03])
|
|
msg_type, payload = parse_frame(raw)
|
|
assert msg_type == MsgType.MCU_READY
|
|
assert payload == bytes([0x01, 0x03])
|
|
|
|
def test_parse_frame_empty_payload(self):
|
|
raw = bytes([MsgType.I2C_STOP])
|
|
msg_type, payload = parse_frame(raw)
|
|
assert msg_type == MsgType.I2C_STOP
|
|
assert payload == b""
|
|
|
|
def test_parse_frame_too_short(self):
|
|
with pytest.raises(ValueError):
|
|
parse_frame(b"")
|
|
|
|
|
|
class TestStreamDeframer:
|
|
def test_single_complete_frame(self):
|
|
df = StreamDeframer()
|
|
frame = build_frame(MsgType.MCU_READY, bytes([0x01, 0x00]))
|
|
msgs = df.feed(frame)
|
|
assert len(msgs) == 1
|
|
assert msgs[0][0] == MsgType.MCU_READY
|
|
assert msgs[0][1] == bytes([0x01, 0x00])
|
|
|
|
def test_partial_then_complete(self):
|
|
df = StreamDeframer()
|
|
frame = build_frame(MsgType.I2C_WRITE, bytes([0x20, 0x00, 0xAB]))
|
|
mid = len(frame) // 2
|
|
msgs1 = df.feed(frame[:mid])
|
|
assert len(msgs1) == 0
|
|
msgs2 = df.feed(frame[mid:])
|
|
assert len(msgs2) == 1
|
|
assert msgs2[0][0] == MsgType.I2C_WRITE
|
|
|
|
def test_multiple_frames_in_one_chunk(self):
|
|
df = StreamDeframer()
|
|
f1 = build_frame(MsgType.I2C_WRITE, bytes([0x20, 0x00]))
|
|
f2 = build_frame(MsgType.I2C_STOP)
|
|
msgs = df.feed(f1 + f2)
|
|
assert len(msgs) == 2
|
|
assert msgs[0][0] == MsgType.I2C_WRITE
|
|
assert msgs[1][0] == MsgType.I2C_STOP
|
|
|
|
def test_corrupted_frame_skipped(self):
|
|
df = StreamDeframer()
|
|
garbage = bytes([0xFF, 0xFE, 0x00])
|
|
valid = build_frame(MsgType.MCU_READY, bytes([0x01, 0x00]))
|
|
msgs = df.feed(garbage + valid)
|
|
assert len(msgs) == 1
|
|
assert msgs[0][0] == MsgType.MCU_READY
|
|
|
|
def test_empty_feed(self):
|
|
df = StreamDeframer()
|
|
msgs = df.feed(b"")
|
|
assert len(msgs) == 0
|
|
|
|
|
|
import time
|
|
from pm3py.sim.mcu_bridge import McuBridge
|
|
|
|
|
|
class MockSerial:
|
|
"""Mock serial port for testing McuBridge."""
|
|
def __init__(self):
|
|
self._rx_buf = bytearray()
|
|
self._tx_buf = bytearray()
|
|
self.is_open = True
|
|
self.timeout = 0.1
|
|
|
|
def read(self, size=1):
|
|
if not self._rx_buf:
|
|
time.sleep(0.01)
|
|
return b""
|
|
data = bytes(self._rx_buf[:size])
|
|
self._rx_buf = self._rx_buf[size:]
|
|
return data
|
|
|
|
def write(self, data):
|
|
self._tx_buf.extend(data)
|
|
return len(data)
|
|
|
|
def inject(self, data: bytes):
|
|
self._rx_buf.extend(data)
|
|
|
|
def get_sent(self) -> bytes:
|
|
data = bytes(self._tx_buf)
|
|
self._tx_buf.clear()
|
|
return data
|
|
|
|
def close(self):
|
|
self.is_open = False
|
|
|
|
|
|
class TestMcuBridge:
|
|
def test_connect_receives_mcu_ready(self):
|
|
mock = MockSerial()
|
|
bridge = McuBridge(port=mock)
|
|
ready_events = []
|
|
bridge.on_ready = lambda ver, caps: ready_events.append((ver, caps))
|
|
bridge.start()
|
|
mock.inject(build_frame(MsgType.MCU_READY, bytes([0x01, 0x03])))
|
|
time.sleep(0.15)
|
|
bridge.stop()
|
|
assert len(ready_events) == 1
|
|
assert ready_events[0] == (0x01, 0x03)
|
|
|
|
def test_i2c_write_callback(self):
|
|
mock = MockSerial()
|
|
bridge = McuBridge(port=mock)
|
|
writes = []
|
|
bridge.on_i2c_write = lambda addr, data: writes.append((addr, data))
|
|
bridge.start()
|
|
mock.inject(build_frame(MsgType.I2C_WRITE, bytes([0x20, 0x00, 0xDE, 0xAD])))
|
|
time.sleep(0.15)
|
|
bridge.stop()
|
|
assert len(writes) == 1
|
|
assert writes[0] == (0x2000, bytes([0xDE, 0xAD]))
|
|
|
|
def test_i2c_read_req_callback(self):
|
|
mock = MockSerial()
|
|
bridge = McuBridge(port=mock)
|
|
reads = []
|
|
bridge.on_i2c_read_req = lambda addr, length: reads.append((addr, length))
|
|
bridge.start()
|
|
mock.inject(build_frame(MsgType.I2C_READ_REQ, bytes([0x20, 0x00, 0x04])))
|
|
time.sleep(0.15)
|
|
bridge.stop()
|
|
assert len(reads) == 1
|
|
assert reads[0] == (0x2000, 4)
|
|
|
|
def test_send_set_ed(self):
|
|
mock = MockSerial()
|
|
bridge = McuBridge(port=mock)
|
|
bridge.start()
|
|
bridge.send_set_ed(True)
|
|
time.sleep(0.05)
|
|
bridge.stop()
|
|
sent = mock.get_sent()
|
|
assert len(sent) > 0
|
|
assert sent[-1] == 0x00
|
|
msg_type, payload = parse_frame(cobs_decode(sent[:-1]))
|
|
assert msg_type == MsgType.SET_ED
|
|
assert payload == bytes([0x01])
|
|
|
|
def test_send_set_eh_voltage(self):
|
|
mock = MockSerial()
|
|
bridge = McuBridge(port=mock)
|
|
bridge.start()
|
|
bridge.send_set_eh_voltage(2400)
|
|
time.sleep(0.05)
|
|
bridge.stop()
|
|
sent = mock.get_sent()
|
|
assert sent[-1] == 0x00
|
|
msg_type, payload = parse_frame(cobs_decode(sent[:-1]))
|
|
assert msg_type == MsgType.SET_EH_VOLTAGE
|
|
assert payload == bytes([0x09, 0x60])
|
|
|
|
def test_send_write_sram(self):
|
|
mock = MockSerial()
|
|
bridge = McuBridge(port=mock)
|
|
bridge.start()
|
|
bridge.send_write_sram(0x10, bytes([0xAA, 0xBB, 0xCC]))
|
|
time.sleep(0.05)
|
|
bridge.stop()
|
|
sent = mock.get_sent()
|
|
assert sent[-1] == 0x00
|
|
msg_type, payload = parse_frame(cobs_decode(sent[:-1]))
|
|
assert msg_type == MsgType.WRITE_SRAM
|
|
assert payload == bytes([0x10, 0xAA, 0xBB, 0xCC])
|
|
|
|
def test_send_i2c_read_response(self):
|
|
mock = MockSerial()
|
|
bridge = McuBridge(port=mock)
|
|
bridge.start()
|
|
bridge.send_i2c_read_response(bytes([0x01, 0x02, 0x03, 0x04]))
|
|
time.sleep(0.05)
|
|
bridge.stop()
|
|
sent = mock.get_sent()
|
|
assert sent[-1] == 0x00
|
|
msg_type, payload = parse_frame(cobs_decode(sent[:-1]))
|
|
assert msg_type == MsgType.I2C_READ_RESPONSE
|
|
assert payload == bytes([0x01, 0x02, 0x03, 0x04])
|
|
|
|
def test_stop_is_idempotent(self):
|
|
mock = MockSerial()
|
|
bridge = McuBridge(port=mock)
|
|
bridge.start()
|
|
bridge.stop()
|
|
bridge.stop() # should not raise
|
|
|
|
|
|
class TestMcuBridgeOpen:
|
|
def test_open_creates_bridge(self):
|
|
from pm3py.sim import McuBridge as ImportedBridge
|
|
assert ImportedBridge is McuBridge
|
|
assert hasattr(McuBridge, 'open')
|