From 980e100a7be6940c2bf4648c461f7ee6963973a3 Mon Sep 17 00:00:00 2001 From: Carlo Perassi Date: Fri, 30 Oct 2026 18:42:48 +0100 Subject: [PATCH] Answer ILLEGAL_VALUE for a Write Single Coil value other than 0x0000/0xFF00 The specification (6.5) allows only 0xFF00 (ON) and 0x0000 (OFF); any other value is illegal, must not affect the coil, and gets exception 03. The server took any non-zero value as ON and echoed 0xFF00. Co-Authored-By: Claude Opus 5.5 --- pymodbus/pdu/bit_message.py | 7 +++++-- test/pdu/test_bit.py | 26 ++++++++++++++++++++++++++ test/server/test_server_asyncio.py | 8 ++++++++ 3 files changed, 39 insertions(+), 2 deletions(-) diff --git a/pymodbus/pdu/bit_message.py b/pymodbus/pdu/bit_message.py index 296e21b5f..49bb68f0e 100644 --- a/pymodbus/pdu/bit_message.py +++ b/pymodbus/pdu/bit_message.py @@ -156,6 +156,7 @@ class WriteSingleCoilRequest(ModbusPDU): """WriteSingleCoilRequest.""" function_code = 5 + _value: int | None = None @classmethod def x_calculateRtuFrameSize(cls, data: bytes) -> int: @@ -170,13 +171,15 @@ def x_encode(self) -> bytes: def x_decode(self, data: bytes) -> None: """Decode a write coil request.""" - self.address, value = struct.unpack(">HH", data[:4]) - self.bits = [bool(value)] + self.address, self._value = struct.unpack(">HH", data[:4]) + self.bits = [bool(self._value)] async def datastore_update( self, context: ModbusServerContext, device_id: int ) -> ModbusPDU: """Run a request against a datastore.""" + if self._value not in (None, ModbusStatus.ON, ModbusStatus.OFF): + return ExceptionResponse(self.function_code, ExcCodes.ILLEGAL_VALUE) if rc := await context.async_setValues( device_id, self.function_code, self.address, self.bits ): diff --git a/test/pdu/test_bit.py b/test/pdu/test_bit.py index 0eb08e0af..ff8748b98 100644 --- a/test/pdu/test_bit.py +++ b/test/pdu/test_bit.py @@ -231,6 +231,32 @@ async def test_write_single_coil_datastore_update(self, mock_server_context): result = await request.datastore_update(context, 0) assert result.x_encode() == b"\x00\x02\x00\x00" + @pytest.mark.parametrize("value", [b"\x00\x01", b"\x00\xff", b"\xff\xff"]) + async def test_write_single_coil_illegal_value(self, value, mock_server_context): + """A value other than 0x0000 or 0xFF00 must answer 0x03 and leave the coil.""" + context = mock_server_context() + context.async_setValues = mock.AsyncMock(return_value=0) + request = bit_msg.WriteSingleCoilRequest() + request.x_decode(b"\x00\x02" + value) + result = await request.datastore_update(context, 1) + assert result.function_code == 0x85 + assert result.exception_code == ExcCodes.ILLEGAL_VALUE + context.async_setValues.assert_not_awaited() + + @pytest.mark.parametrize( + ("value", "bit"), [(b"\xff\x00", True), (b"\x00\x00", False)] + ) + async def test_write_single_coil_legal_value(self, value, bit, mock_server_context): + """0xFF00 and 0x0000 are written as ON and OFF.""" + context = mock_server_context() + context.async_setValues = mock.AsyncMock(return_value=0) + context.async_getValues = mock.AsyncMock(return_value=[bit]) + request = bit_msg.WriteSingleCoilRequest() + request.x_decode(b"\x00\x02" + value) + result = await request.datastore_update(context, 1) + assert result.x_encode() == b"\x00\x02" + value + context.async_setValues.assert_awaited_once_with(1, 5, 2, [bit]) + async def test_write_multiple_coils_datastore_update(self, mock_server_context): """Test write multiple coils.""" context = mock_server_context(False) diff --git a/test/server/test_server_asyncio.py b/test/server/test_server_asyncio.py index bea97b927..4072109b3 100755 --- a/test/server/test_server_asyncio.py +++ b/test/server/test_server_asyncio.py @@ -266,6 +266,14 @@ async def test_async_tcp_server_read_count_above_limit(self): await asyncio.wait_for(BasicClient.done, timeout=0.1) assert BasicClient.received_data == b"\x00\x00\x00\x00\x00\x03\x01\x83\x03" + async def test_async_tcp_server_write_coil_illegal_value(self): + """A Write Single Coil value other than 0x0000 or 0xFF00 must answer 0x03.""" + BasicClient.data = b"\x00\x00\x00\x00\x00\x06\x01\x05\x00\x00\x00\x01" + await self.start_server() + await self.connect_server() + await asyncio.wait_for(BasicClient.done, timeout=0.1) + assert BasicClient.received_data == b"\x00\x00\x00\x00\x00\x03\x01\x85\x03" + async def test_async_server_trace_connect_disconnect(self): """Test connect/disconnect trace handler.""" trace_connect = mock.Mock()