"""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