70 lines
1.9 KiB
Python
70 lines
1.9 KiB
Python
"""MBAP codec tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from wago_sim.mbap import (
|
|
MBAP_HEADER_SIZE,
|
|
MbapError,
|
|
MbapFrameReader,
|
|
MbapHeader,
|
|
build_frame,
|
|
)
|
|
|
|
|
|
def test_header_roundtrip():
|
|
header = MbapHeader(transaction_id=0x1234, protocol_id=0, length=6, unit_id=1)
|
|
encoded = header.encode()
|
|
assert len(encoded) == 7
|
|
decoded = MbapHeader.decode(encoded + b"\x03\x00\x00\x00\x02")
|
|
assert decoded == header
|
|
|
|
|
|
def test_decode_requires_complete_frame():
|
|
with pytest.raises(MbapError):
|
|
MbapHeader.decode(b"\x00\x01\x00\x00\x00\x06\x01")
|
|
|
|
|
|
def test_protocol_id_rejected():
|
|
with pytest.raises(MbapError):
|
|
MbapHeader.decode(b"\x00\x01\x00\x01\x00\x06\x01\x03\x00\x00\x00\x02")
|
|
|
|
|
|
def test_build_frame():
|
|
frame = build_frame(7, 1, b"\x03\x00\x00\x00\x02")
|
|
assert frame[:2] == (7).to_bytes(2, "big")
|
|
assert frame[4:6] == (6).to_bytes(2, "big")
|
|
assert frame[6] == 1
|
|
|
|
|
|
def test_frame_reader_splits_stream():
|
|
reader = MbapFrameReader()
|
|
f1 = build_frame(1, 1, b"\x03\x00\x00\x00\x02")
|
|
f2 = build_frame(2, 1, b"\x01\x00\x00\x00\x08")
|
|
blob = f1 + f2
|
|
# feed an incomplete header first
|
|
reader.feed(blob[:5])
|
|
assert reader.pop_frame() is None
|
|
# feed through the end of frame 1 (12 bytes): it becomes complete
|
|
reader.feed(blob[5:12])
|
|
h1, p1 = reader.pop_frame()
|
|
assert h1.transaction_id == 1
|
|
assert p1 == b"\x03\x00\x00\x00\x02"
|
|
# frame 2 arrives in pieces
|
|
reader.feed(blob[12:15])
|
|
assert reader.pop_frame() is None
|
|
reader.feed(blob[15:])
|
|
h2, p2 = reader.pop_frame()
|
|
assert h2.transaction_id == 2
|
|
assert p2 == b"\x01\x00\x00\x00\x08"
|
|
assert reader.pop_frame() is None
|
|
|
|
|
|
def test_frame_reader_rejects_bad_protocol():
|
|
reader = MbapFrameReader()
|
|
reader.feed(b"\x00\x01\xDE\xAD\x00\x06\x01\x03\x00\x00\x00\x02")
|
|
import pytest
|
|
|
|
with pytest.raises(MbapError):
|
|
reader.pop_frame() |