Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 9 additions & 2 deletions pymodbus/pdu/other_message.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@

from ..constants import ModbusStatus
from ..datastore import ModbusServerContext
from ..exceptions import ModbusIOException
from .decoders import DecodePDU
from .device import DeviceInformationFactory, ModbusControlBlock
from .pdu import ModbusPDU
Expand Down Expand Up @@ -243,8 +244,14 @@ def decode(self, data: bytes) -> None:
raw value that a user can decode to whatever it should be.
"""
self.byte_count = int(data[0])
self.identifier = data[1 : self.byte_count + 1]
status = int(data[-1])
if not 1 <= self.byte_count <= len(data) - 1:
raise ModbusIOException(
f"byte_count {self.byte_count} outside 1..{len(data) - 1} "
f"for packet of length {len(data)}",
function_code=self.function_code,
)
self.identifier = data[1 : self.byte_count]
status = int(data[self.byte_count])
self.status = status == ID_ON


Expand Down
24 changes: 23 additions & 1 deletion test/pdu/test_other_messages.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,10 @@
from typing import cast
from unittest import mock

import pytest

import pymodbus.pdu.other_message as pymodbus_message
from pymodbus.exceptions import ModbusIOException


class TestOtherMessage:
Expand Down Expand Up @@ -155,9 +158,28 @@ async def test_report_device_id(self, mock_server_context):
)

assert response.encode() == b"\tPymodbus\xff"
response.decode(b"\x03\x12\x00")
response.decode(b"\tPymodbus\xff")
assert response.status
assert response.identifier == b"Pymodbus"
response.decode(b"\x03\x12\x00\x00")
assert not response.status
assert response.identifier == b"\x12\x00"

response.status = False
assert response.encode() == b"\x03\x12\x00\x00"

def test_report_device_id_response_roundtrip(self):
"""Test decode(encode()) keeps identifier and status unchanged."""
for identifier, status in ((b"Pymodbus", True), (b"\x12", False)):
sent = pymodbus_message.ReportDeviceIdResponse(identifier, status)
received = pymodbus_message.ReportDeviceIdResponse()
received.decode(sent.encode())
assert received.identifier == identifier
assert received.status == status

def test_report_device_id_response_invalid_byte_count(self):
"""Test byte count pointing outside the received data raises."""
response = pymodbus_message.ReportDeviceIdResponse()
for frame in (b"\x00\x12\xff", b"\x04\x12\xff"):
with pytest.raises(ModbusIOException):
response.decode(frame)