Answer ILLEGAL_VALUE for a Write Single Coil value other than 0x0000/0xFF00

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
         ):

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)

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()