188 lines
6.9 KiB
Python
188 lines
6.9 KiB
Python
"""PDU codec tests: request decoding, response encoding, limits."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from wago_sim.pdu import (
|
|
COIL_OFF,
|
|
COIL_ON,
|
|
EXC_ILLEGAL_DATA_ADDRESS,
|
|
EXC_ILLEGAL_DATA_VALUE,
|
|
ExceptionResponse,
|
|
GetCommEventCountResponse,
|
|
IllegalDataValue,
|
|
MaskWriteRegisterResponse,
|
|
ReadRequest,
|
|
ReadResponse,
|
|
UnsupportedFunction,
|
|
WriteMultipleResponse,
|
|
WriteSingleCoilRequest,
|
|
WriteSingleCoilResponse,
|
|
WriteSingleRegisterRequest,
|
|
WriteSingleRegisterResponse,
|
|
decode_request,
|
|
FC_GET_COMM_EVENT_COUNT,
|
|
FC_MASK_WRITE_REGISTER,
|
|
FC_READ_COILS,
|
|
FC_READ_DISCRETE_INPUTS,
|
|
FC_READ_HOLDING_REGISTERS,
|
|
FC_READ_INPUT_REGISTERS,
|
|
FC_READ_WRITE_MULTIPLE_REGISTERS,
|
|
FC_WRITE_MULTIPLE_COILS,
|
|
FC_WRITE_MULTIPLE_REGISTERS,
|
|
FC_WRITE_SINGLE_COIL,
|
|
FC_WRITE_SINGLE_REGISTER,
|
|
MAX_READ_BITS,
|
|
MAX_READ_REGISTERS,
|
|
MAX_WRITE_BITS,
|
|
MAX_WRITE_REGISTERS,
|
|
)
|
|
|
|
|
|
def _req(fc: int, body: bytes) -> bytes:
|
|
return bytes([fc]) + body
|
|
|
|
|
|
class TestReadRequests:
|
|
@pytest.mark.parametrize("fc,max_count", [
|
|
(FC_READ_COILS, MAX_READ_BITS),
|
|
(FC_READ_DISCRETE_INPUTS, MAX_READ_BITS),
|
|
(FC_READ_HOLDING_REGISTERS, MAX_READ_REGISTERS),
|
|
(FC_READ_INPUT_REGISTERS, MAX_READ_REGISTERS),
|
|
])
|
|
def test_read_ok(self, fc, max_count):
|
|
pdu = _req(fc, (0x00).to_bytes(2, "big") + max_count.to_bytes(2, "big"))
|
|
req = decode_request(pdu)
|
|
assert isinstance(req, ReadRequest)
|
|
assert req.function_code == fc
|
|
assert req.count == max_count
|
|
|
|
def test_read_count_zero_rejected(self):
|
|
with pytest.raises(IllegalDataValue):
|
|
decode_request(_req(FC_READ_COILS, b"\x00\x00\x00\x00"))
|
|
|
|
def test_read_count_over_limit_rejected(self):
|
|
with pytest.raises(IllegalDataValue):
|
|
decode_request(_req(FC_READ_COILS, b"\x00\x00\x07\xD1")) # 2001
|
|
with pytest.raises(IllegalDataValue):
|
|
decode_request(_req(FC_READ_HOLDING_REGISTERS, b"\x00\x00\x00\x7E")) # 126
|
|
|
|
def test_read_body_length(self):
|
|
with pytest.raises(Exception):
|
|
decode_request(_req(FC_READ_COILS, b"\x00\x00\x00"))
|
|
|
|
|
|
class TestWriteRequests:
|
|
def test_write_single_coil_on(self):
|
|
req = decode_request(_req(FC_WRITE_SINGLE_COIL, b"\x00\x05\xFF\x00"))
|
|
assert isinstance(req, WriteSingleCoilRequest)
|
|
assert req.address == 5
|
|
assert req.value is True
|
|
|
|
def test_write_single_coil_bad_value(self):
|
|
with pytest.raises(IllegalDataValue):
|
|
decode_request(_req(FC_WRITE_SINGLE_COIL, b"\x00\x05\x12\x34"))
|
|
|
|
def test_write_single_register(self):
|
|
req = decode_request(_req(FC_WRITE_SINGLE_REGISTER, b"\x00\x0A\x12\x34"))
|
|
assert isinstance(req, WriteSingleRegisterRequest)
|
|
assert req.value == 0x1234
|
|
|
|
def test_write_multiple_coils(self):
|
|
# address 0, 10 coils, values 0b0000_0001_01
|
|
body = b"\x00\x00\x00\x0A\x02\x05\x01"
|
|
req = decode_request(_req(FC_WRITE_MULTIPLE_COILS, body))
|
|
assert req.function_code == FC_WRITE_MULTIPLE_COILS
|
|
assert len(req.values) == 10
|
|
assert req.values == (True, False, True, False, False, False, False,
|
|
False, True, False)
|
|
|
|
def test_write_multiple_coils_count_mismatch(self):
|
|
body = b"\x00\x00\x00\x0A\x03\x05\x01\x00" # announces 3 bytes for 10 bits
|
|
with pytest.raises(Exception):
|
|
decode_request(_req(FC_WRITE_MULTIPLE_COILS, body))
|
|
|
|
def test_write_multiple_coils_over_limit(self):
|
|
body = b"\x00\x00\x03\x21\x65" + b"\xFF" * 0x65
|
|
with pytest.raises(IllegalDataValue):
|
|
decode_request(_req(FC_WRITE_MULTIPLE_COILS, body))
|
|
|
|
def test_write_multiple_registers(self):
|
|
body = b"\x00\x00\x00\x02\x04\x12\x34\x56\x78"
|
|
req = decode_request(_req(FC_WRITE_MULTIPLE_REGISTERS, body))
|
|
assert req.values == (0x1234, 0x5678)
|
|
|
|
def test_write_multiple_registers_over_limit(self):
|
|
count = MAX_WRITE_REGISTERS + 1
|
|
body = count.to_bytes(2, "big")[:2] + (2 * count).to_bytes(1, "big") + b"\x00" * (2 * count)
|
|
# count field is 2 bytes; re-encode properly
|
|
body = b"\x00\x00" + count.to_bytes(2, "big") + (2 * count).to_bytes(1, "big") + b"\x00" * (2 * count)
|
|
with pytest.raises(IllegalDataValue):
|
|
decode_request(_req(FC_WRITE_MULTIPLE_REGISTERS, body))
|
|
|
|
def test_mask_write(self):
|
|
req = decode_request(_req(FC_MASK_WRITE_REGISTER, b"\x00\x0A\x00\xF2\x00\x25"))
|
|
assert req.address == 0x0A
|
|
assert req.and_mask == 0x00F2
|
|
assert req.or_mask == 0x0025
|
|
|
|
def test_read_write_multiple(self):
|
|
body = (b"\x00\x00\x00\x02" # read addr 0, count 2
|
|
b"\x10\x00\x00\x02" # write addr 0x1000, count 2
|
|
b"\x04\xAA\xBB\xCC\xDD") # byte count (1 byte) + 4 data bytes
|
|
req = decode_request(_req(FC_READ_WRITE_MULTIPLE_REGISTERS, body))
|
|
assert req.read_count == 2
|
|
assert req.write_address == 0x1000
|
|
assert req.values == (0xAABB, 0xCCDD)
|
|
|
|
|
|
class TestUnsupported:
|
|
def test_fc7_rejected(self):
|
|
with pytest.raises(UnsupportedFunction) as ei:
|
|
decode_request(_req(0x07, b"\x00\x00"))
|
|
assert ei.value.function_code == 0x07
|
|
|
|
def test_fc8_rejected(self):
|
|
with pytest.raises(UnsupportedFunction):
|
|
decode_request(_req(0x08, b"\x00\x00\x00\x00"))
|
|
|
|
|
|
class TestResponses:
|
|
def test_read_bits_encoding(self):
|
|
resp = ReadResponse(FC_READ_COILS)
|
|
resp.bits = [True, False, True] + [False] * 5
|
|
assert resp.encode() == b"\x01\x01\x05"
|
|
|
|
def test_read_registers_encoding(self):
|
|
resp = ReadResponse(FC_READ_HOLDING_REGISTERS)
|
|
resp.registers = [0x1234, 0x5678]
|
|
assert resp.encode() == b"\x03\x04\x12\x34\x56\x78"
|
|
|
|
def test_write_single_coil_response(self):
|
|
resp = WriteSingleCoilResponse(5, True)
|
|
assert resp.encode() == b"\x05\x00\x05\xFF\x00"
|
|
|
|
def test_write_single_register_response(self):
|
|
resp = WriteSingleRegisterResponse(2, 0xBEEF)
|
|
assert resp.encode() == b"\x06\x00\x02\xBE\xEF"
|
|
|
|
def test_write_multiple_response(self):
|
|
resp = WriteMultipleResponse(FC_WRITE_MULTIPLE_REGISTERS, 0, 4)
|
|
assert resp.encode() == b"\x10\x00\x00\x00\x04"
|
|
|
|
def test_mask_write_response(self):
|
|
resp = MaskWriteRegisterResponse(4, 0x00F2, 0x0025)
|
|
assert resp.encode() == b"\x16\x00\x04\x00\xF2\x00\x25"
|
|
|
|
def test_event_count_response(self):
|
|
resp = GetCommEventCountResponse(0, 42)
|
|
assert resp.encode() == b"\x0B\x00\x00\x00\x2A"
|
|
|
|
def test_exception_response(self):
|
|
resp = ExceptionResponse(FC_READ_COILS, EXC_ILLEGAL_DATA_ADDRESS)
|
|
assert resp.encode() == b"\x81\x02"
|
|
|
|
def test_event_count_request(self):
|
|
req = decode_request(_req(FC_GET_COMM_EVENT_COUNT, b""))
|
|
assert req.function_code == FC_GET_COMM_EVENT_COUNT |