pymodbus-dev__pymodbus-1923

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "",
  "details": {
    "stdout": "",
    "stderr": "",
    "exit_code": 1
  }
}

{
  "status": "ok",
  "filepath": "pymodbus/framer/socket_framer.py",
  "content": "\"\"\"Socket framer.\"\"\"\n# pylint: disable=missing-type-doc\nimport struct\n\nfrom pymodbus.exceptions import (\n    InvalidMessageReceivedException,\n    ModbusIOException,\n)\nfrom pymodbus.framer.base import SOCKET_FRAME_HEADER, ModbusFramer\nfrom pymodbus.logging import Log\n\n\n# --------------------------------------------------------------------------- #\n# Modbus TCP Message\n# --------------------------------------------------------------------------- #\n\n\nclass ModbusSocketFramer(ModbusFramer):\n    \"\"\"Modbus Socket Frame controller.\n\n    Before each modbus TCP message is an MBAP header which is used as a\n    message frame.  It allows us to easily separate messages as follows::\n\n        [         MBAP Header         ] [ Function Code] [ Data ] \\\n        [ tid ][ pid ][ length ][ uid ]\n          2b     2b     2b        1b           1b           Nb\n\n        while len(message) > 0:\n            tid, pid, length`, uid = struct.unpack(\">HHHB\", message)\n            request = message[0:7 + length - 1`]\n            message = [7 + length - 1:]\n\n        * length = uid + function code + data\n        * The -1 is to account for the uid byte\n    \"\"\"\n\n    method = \"socket\"\n\n    def __init__(self, decoder, client=None):\n        \"\"\"Initialize a new instance of the framer.\n\n        :param decoder: The decoder factory implementation to use\n        \"\"\"\n        super().__init__(decoder, client)\n        self._hsize = 0x07\n\n    # ----------------------------------------------------------------------- #\n    # Private Helper Functions\n    # ----------------------------------------------------------------------- #\n    def checkFrame(self):\n        \"\"\"Check and decode the next frame.\n\n        Return true if we were successful.\n        \"\"\"\n        if self.isFrameReady():\n            (\n                self._header[\"tid\"],\n                self._header[\"pid\"],\n                self._header[\"len\"],\n                self._header[\"uid\"],\n            ) = struct.unpack(\">HHHB\", self._buffer[0 : self._hsize])\n\n            # someone sent us an error? ignore it\n            if self._header[\"len\"] < 2:\n                self.advanceFrame()\n            # we have at least a complete message, continue\n            elif len(self._buffer) - self._hsize + 1 >= self._header[\"len\"]:\n                return True\n        # we don't have enough of a message yet, wait\n        return False\n\n    def advanceFrame(self):\n        \"\"\"Skip over the current framed message.\n\n        This allows us to skip over the current message after we have processed\n        it or determined that it contains an error. It also has to reset the\n        current frame header handle\n        \"\"\"\n        length = self._hsize + self._header[\"len\"] - 1\n        self._buffer = self._buffer[length:]\n        self._header = {\"tid\": 0, \"pid\": 0, \"len\": 0, \"uid\": 0}\n\n    def isFrameReady(self):\n        \"\"\"Check if we should continue decode logic.\n\n        This is meant to be used in a while loop in the decoding phase to let\n        the decoder factory know that there is still data in the buffer.\n\n        :returns: True if ready, False otherwise\n        \"\"\"\n        return len(self._buffer) > self._hsize\n\n    def getFrame(self):\n        \"\"\"Return the next frame from the buffered data.\n\n        :returns: The next full frame buffer\n        \"\"\"\n        length = self._hsize + self._header[\"len\"] - 1\n        return self._buffer[self._hsize : length]\n\n    # ----------------------------------------------------------------------- #\n    # Public Member Functions\n    # ----------------------------------------------------------------------- #\n    def decode_data(self, data):\n        \"\"\"Decode data.\"\"\"\n        if len(data) > self._hsize:\n            tid, pid, length, uid, fcode = struct.unpack(\n                SOCKET_FRAME_HEADER, data[0 : self._hsize + 1]\n            )\n            return {\n                \"tid\": tid,\n                \"pid\": pid,\n                \"length\": length,\n                \"slave\": uid,\n                \"fcode\": fcode,\n            }\n        return {}\n\n    def frameProcessIncomingPacket(self, single, callback, slave, tid=None, **kwargs):\n        \"\"\"Process new packet pattern.\n\n        This takes in a new request packet, adds it to the current\n        packet stream, and performs framing on it. That is, checks\n        for complete messages, and once found, will process all that\n        exist.  This handles the case when we read N + 1 or 1 // N\n        messages at a time instead of 1.\n\n        The processed and decoded messages are pushed to the callback\n        function to process and send.\n        \"\"\"\n        while True:\n            if not self.isFrameReady():\n                if len(self._buffer):\n                    # Possible error ???\n                    if self._header[\"len\"] < 2:\n                        self._process(callback, tid, error=True)\n                break\n            if not self.checkFrame():\n                Log.debug(\"Frame check failed, ignoring!!\")\n                self.resetFrame()\n                continue\n            if not self._validate_slave_id(slave, single):\n                header_txt = self._header[\"uid\"]\n                Log.debug(\"Not a valid slave id - {}, ignoring!!\", header_txt)\n                self.resetFrame()\n                continue\n            self._process(callback, tid)\n\n    def _process(self, callback, tid, error=False):\n        \"\"\"Process incoming packets irrespective error condition.\"\"\"",
  "start_line": 1,
  "end_line": 150,
  "total_lines": 182,
  "is_truncated": true
}

{
  "status": "ok",
  "filepath": "pymodbus/framer/socket_framer.py",
  "content": "        data = self._buffer if error else self.getFrame()\n        if (result := self.decoder.decode(data)) is None:\n            self.resetFrame()\n            raise ModbusIOException(\"Unable to decode request\")\n        if error and result.function_code < 0x80:\n            raise InvalidMessageReceivedException(result)\n        self.populateResult(result)\n        self.advanceFrame()\n        if tid and tid != result.transaction_id:\n            self.resetFrame()\n        else:\n            callback(result)  # defer or push to a thread?\n\n    def buildPacket(self, message):\n        \"\"\"Create a ready to send modbus packet.\n\n        :param message: The populated request/response to send\n        \"\"\"\n        data = message.encode()\n        packet = struct.pack(\n            SOCKET_FRAME_HEADER,\n            message.transaction_id,\n            message.protocol_id,\n            len(data) + 2,\n            message.slave_id,\n            message.function_code,\n        )\n        packet += data\n        return packet\n\n\n# __END__",
  "start_line": 151,
  "end_line": 182,
  "total_lines": 182,
  "is_truncated": false
}

{
  "status": "ok",
  "filepath": "pymodbus/framer/base.py",
  "content": "\"\"\"Framer start.\"\"\"\n# pylint: disable=missing-type-doc\nfrom __future__ import annotations\n\nfrom typing import Any\n\nfrom pymodbus.factory import ClientDecoder, ServerDecoder\nfrom pymodbus.logging import Log\n\n\n# Unit ID, Function Code\nBYTE_ORDER = \">\"\nFRAME_HEADER = \"BB\"\n\n# Transaction Id, Protocol ID, Length, Unit ID, Function Code\nSOCKET_FRAME_HEADER = BYTE_ORDER + \"HHH\" + FRAME_HEADER\n\n# Function Code\nTLS_FRAME_HEADER = BYTE_ORDER + \"B\"\n\n\nclass ModbusFramer:\n    \"\"\"Base Framer class.\"\"\"\n\n    name = \"\"\n\n    def __init__(\n        self,\n        decoder: ClientDecoder | ServerDecoder,\n        client=None,\n    ) -> None:\n        \"\"\"Initialize a new instance of the framer.\n\n        :param decoder: The decoder implementation to use\n        \"\"\"\n        self.decoder = decoder\n        self.client = client\n        self._header: dict[str, Any] = {\n            \"lrc\": \"0000\",\n            \"len\": 0,\n            \"uid\": 0x00,\n            \"tid\": 0,\n            \"pid\": 0,\n            \"crc\": b\"\\x00\\x00\",\n        }\n        self._buffer = b\"\"\n\n    def _validate_slave_id(self, slaves: list, single: bool) -> bool:\n        \"\"\"Validate if the received data is valid for the client.\n\n        :param slaves: list of slave id for which the transaction is valid\n        :param single: Set to true to treat this as a single context\n        :return:\n        \"\"\"\n        if single:\n            return True\n        if 0 in slaves or 0xFF in slaves:\n            # Handle Modbus TCP slave identifier (0x00 0r 0xFF)\n            # in asynchronous requests\n            return True\n        return self._header[\"uid\"] in slaves\n\n    def sendPacket(self, message):\n        \"\"\"Send packets on the bus.\n\n        With 3.5char delay between frames\n        :param message: Message to be sent over the bus\n        :return:\n        \"\"\"\n        return self.client.send(message)\n\n    def recvPacket(self, size):\n        \"\"\"Receive packet from the bus.\n\n        With specified len\n        :param size: Number of bytes to read\n        :return:\n        \"\"\"\n        return self.client.recv(size)\n\n    def resetFrame(self):\n        \"\"\"Reset the entire message frame.\n\n        This allows us to skip ovver errors that may be in the stream.\n        It is hard to know if we are simply out of sync or if there is\n        an error in the stream as we have no way to check the start or\n        end of the message (python just doesn't have the resolution to\n        check for millisecond delays).\n        \"\"\"\n        Log.debug(\n            \"Resetting frame - Current Frame in buffer - {}\", self._buffer, \":hex\"\n        )\n        self._buffer = b\"\"\n        self._header = {\n            \"lrc\": \"0000\",\n            \"crc\": b\"\\x00\\x00\",\n            \"len\": 0,\n            \"uid\": 0x00,\n            \"pid\": 0,\n            \"tid\": 0,\n        }\n\n    def populateResult(self, result):\n        \"\"\"Populate the modbus result header.\n\n        The serial packets do not have any header information\n        that is copied.\n\n        :param result: The response packet\n        \"\"\"\n        result.slave_id = self._header.get(\"uid\", 0)\n        result.transaction_id = self._header.get(\"tid\", 0)\n        result.protocol_id = self._header.get(\"pid\", 0)\n\n    def processIncomingPacket(self, data, callback, slave, **kwargs):\n        \"\"\"Process new packet pattern.\n\n        This takes in a new request packet, adds it to the current\n        packet stream, and performs framing on it. That is, checks\n        for complete messages, and once found, will process all that\n        exist.  This handles the case when we read N + 1 or 1 // N\n        messages at a time instead of 1.\n\n        The processed and decoded messages are pushed to the callback\n        function to process and send.\n\n        :param data: The new packet data\n        :param callback: The function to send results to\n        :param slave: Process if slave id matches, ignore otherwise (could be a\n               list of slave ids (server) or single slave id(client/server))\n        :param kwargs:\n        :raises ModbusIOException:\n        \"\"\"\n        Log.debug(\"Processing: {}\", data, \":hex\")\n        self._buffer += data\n        if not isinstance(slave, (list, tuple)):\n            slave = [slave]\n        single = kwargs.pop(\"single\", False)\n        self.frameProcessIncomingPacket(single, callback, slave, **kwargs)\n\n    def frameProcessIncomingPacket(\n        self, _single, _callback, _slave, _tid=None, **kwargs\n    ):\n        \"\"\"Process new packet pattern.\"\"\"",
  "start_line": 1,
  "end_line": 144,
  "total_lines": 144,
  "is_truncated": false
}

{
  "status": "ok",
  "filepath": "test/test_framers.py",
  "content": "\"\"\"Test framers.\"\"\"\nfrom unittest import mock\n\nimport pytest\n\nfrom pymodbus import Framer\nfrom pymodbus.bit_read_message import ReadCoilsRequest\nfrom pymodbus.client.base import ModbusBaseClient\nfrom pymodbus.exceptions import ModbusIOException\nfrom pymodbus.factory import ClientDecoder\nfrom pymodbus.framer.ascii_framer import ModbusAsciiFramer\nfrom pymodbus.framer.binary_framer import ModbusBinaryFramer\nfrom pymodbus.framer.rtu_framer import ModbusRtuFramer\nfrom pymodbus.transport import CommType\nfrom pymodbus.utilities import ModbusTransactionState\n\n\nBASE_PORT = 6600\n\n\nTEST_MESSAGE = b\"\\x00\\x01\\x00\\x01\\x00\\n\\xec\\x1c\"\n\n\n@pytest.fixture(name=\"rtu_framer\")\ndef fixture_rtu_framer():\n    \"\"\"RTU framer.\"\"\"\n    return ModbusRtuFramer(ClientDecoder())\n\n\n@pytest.fixture(name=\"ascii_framer\")\ndef fixture_ascii_framer():\n    \"\"\"Ascii framer.\"\"\"\n    return ModbusAsciiFramer(ClientDecoder())\n\n\n# @pytest.fixture\n# def binary_framer():\n#     \"\"\"Binary framer.\"\"\"\n#     return ModbusBinaryFramer(ClientDecoder())\n\n\n@pytest.mark.parametrize(\n    \"framer\",\n    [\n        ModbusRtuFramer,\n        ModbusAsciiFramer,\n        ModbusBinaryFramer,\n    ],\n)\ndef test_framer_initialization(framer):\n    \"\"\"Test framer initialization.\"\"\"\n    decoder = ClientDecoder()\n    framer = framer(decoder)\n    assert framer.client is None\n    assert framer._buffer == b\"\"  # pylint: disable=protected-access\n    assert framer.decoder == decoder\n    if isinstance(framer, ModbusAsciiFramer):\n        assert framer._header == {  # pylint: disable=protected-access\n            \"tid\": 0,\n            \"pid\": 0,\n            \"lrc\": \"0000\",\n            \"len\": 0,\n            \"uid\": 0x00,\n            \"crc\": b\"\\x00\\x00\",\n        }\n        assert framer._hsize == 0x02  # pylint: disable=protected-access\n        assert framer._start == b\":\"  # pylint: disable=protected-access\n        assert framer._end == b\"\\r\\n\"  # pylint: disable=protected-access\n    elif isinstance(framer, ModbusRtuFramer):\n        assert framer._header == {  # pylint: disable=protected-access\n            \"tid\": 0,\n            \"pid\": 0,\n            \"lrc\": \"0000\",\n            \"uid\": 0x00,\n            \"len\": 0,\n            \"crc\": b\"\\x00\\x00\",\n        }\n        assert framer._hsize == 0x01  # pylint: disable=protected-access\n        assert framer._end == b\"\\x0d\\x0a\"  # pylint: disable=protected-access\n        assert framer._min_frame_size == 4  # pylint: disable=protected-access\n    else:\n        assert framer._header == {  # pylint: disable=protected-access\n            \"tid\": 0,\n            \"pid\": 0,\n            \"lrc\": \"0000\",\n            \"crc\": b\"\\x00\\x00\",\n            \"len\": 0,\n            \"uid\": 0x00,\n        }\n        assert framer._hsize == 0x01  # pylint: disable=protected-access\n        assert framer._start == b\"\\x7b\"  # pylint: disable=protected-access\n        assert framer._end == b\"\\x7d\"  # pylint: disable=protected-access\n        assert framer._repeat == [  # pylint: disable=protected-access\n            b\"}\"[0],\n            b\"{\"[0],\n        ]\n\n\n@pytest.mark.parametrize(\"data\", [(b\"\", {}), (b\"abcd\", {\"fcode\": 98, \"slave\": 97})])\ndef test_decode_data(rtu_framer, data):\n    \"\"\"Test decode data.\"\"\"\n    data, expected = data\n    decoded = rtu_framer.decode_data(data)\n    assert decoded == expected\n\n\n@pytest.mark.parametrize(\n    \"data\",\n    [\n        (b\"\", False),\n        (b\"\\x02\\x01\\x01\\x00Q\\xcc\", True),\n        (b\"\\x11\\x03\\x06\\xAE\\x41\\x56\\x52\\x43\\x40\\x49\\xAD\", True),  # valid frame\n        (b\"\\x11\\x03\\x06\\xAE\\x41\\x56\\x52\\x43\\x40\\x49\\xAC\", False),  # invalid frame CRC\n    ],\n)\ndef test_check_frame(rtu_framer, data):\n    \"\"\"Test check frame.\"\"\"\n    data, expected = data\n    rtu_framer._buffer = data  # pylint: disable=protected-access\n    assert expected == rtu_framer.checkFrame()\n\n\n@pytest.mark.parametrize(\n    \"data\",\n    [\n        (b\"\", {\"uid\": 0x00, \"len\": 0, \"crc\": b\"\\x00\\x00\"}, b\"\"),\n        (b\"abcd\", {\"uid\": 0x00, \"len\": 2, \"crc\": b\"\\x00\\x00\"}, b\"cd\"),\n        (\n            b\"\\x11\\x03\\x06\\xAE\\x41\\x56\\x52\\x43\\x40\\x49\\xAD\\x12\\x03\",  # real case, frame size is 11\n            {\"uid\": 0x00, \"len\": 11, \"crc\": b\"\\x00\\x00\"},\n            b\"\\x12\\x03\",\n        ),\n    ],\n)\ndef test_rtu_advance_framer(rtu_framer, data):\n    \"\"\"Test rtu advance framer.\"\"\"\n    before_buf, before_header, after_buf = data\n\n    rtu_framer._buffer = before_buf  # pylint: disable=protected-access\n    rtu_framer._header = before_header  # pylint: disable=protected-access\n    rtu_framer.advanceFrame()\n    assert rtu_framer._header == {  # pylint: disable=protected-access\n        \"uid\": 0x00,\n        \"len\": 0,\n        \"crc\": b\"\\x00\\x00\",\n    }\n    assert rtu_framer._buffer == after_buf  # pylint: disable=protected-access\n\n\n@pytest.mark.parametrize(\"data\", [b\"\", b\"abcd\"])",
  "start_line": 1,
  "end_line": 150,
  "total_lines": 361,
  "is_truncated": true
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "",
  "details": {
    "stdout": "",
    "stderr": "",
    "exit_code": 1
  }
}

{
  "status": "ok",
  "stdout": "Test failed: Modbus Error: [Input/Output] Unable to decode request\n",
  "stderr": "General exception: index out of range\nTraceback (most recent call last):\n  File \"<stdin>\", line 46, in <module>\n  File \"<stdin>\", line 37, in test_socket_framer_partial_packet\n  File \"/workspace/pymodbus/framer/base.py\", line 139, in processIncomingPacket\n    self.frameProcessIncomingPacket(single, callback, slave, **kwargs)\n  File \"/workspace/pymodbus/framer/socket_framer.py\", line 136, in frameProcessIncomingPacket\n    self._process(callback, tid, error=True)\n  File \"/workspace/pymodbus/framer/socket_framer.py\", line 154, in _process\n    raise ModbusIOException(\"Unable to decode request\")\npymodbus.exceptions.ModbusIOException: Modbus Error: [Input/Output] Unable to decode request\n",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "pymodbus/framer/socket_framer.py",
  "occurrences": 1,
  "strategy": "exact",
  "diff": "--- a/pymodbus/framer/socket_framer.py\n+++ b/pymodbus/framer/socket_framer.py\n@@ -136,9 +136,7 @@\n                         self._process(callback, tid, error=True)\n                 break\n             if not self.checkFrame():\n-                Log.debug(\"Frame check failed, ignoring!!\")\n-                self.resetFrame()\n-                continue\n+                break\n             if not self._validate_slave_id(slave, single):\n                 header_txt = self._header[\"uid\"]\n                 Log.debug(\"Not a valid slave id - {}, ignoring!!\", header_txt)\n",
  "is_truncated": false
}