From 975809c94026ac2327ad06ee9bed57333bc32b0d Mon Sep 17 00:00:00 2001 From: jan iversen Date: Sun, 17 Nov 2024 15:06:30 +0100 Subject: [PATCH] Extend TransactionManager to handle sync. (#2457) --- pymodbus/client/base.py | 21 +-- pymodbus/client/serial.py | 2 + pymodbus/client/tcp.py | 3 +- pymodbus/client/udp.py | 3 +- pymodbus/transaction/__init__.py | 6 - pymodbus/transaction/old_transaction.py | 213 ----------------------- pymodbus/transaction/transaction.py | 84 +++++++-- pymodbus/transport/transport.py | 7 +- test/sub_client/test_client.py | 4 +- test/transaction/test_old_transaction.py | 123 ------------- test/transaction/test_transaction.py | 125 +++++++++++-- 11 files changed, 209 insertions(+), 382 deletions(-) delete mode 100644 pymodbus/transaction/old_transaction.py delete mode 100755 test/transaction/test_old_transaction.py diff --git a/pymodbus/client/base.py b/pymodbus/client/base.py index 95f4f8160..1ab97b33f 100644 --- a/pymodbus/client/base.py +++ b/pymodbus/client/base.py @@ -11,7 +11,7 @@ from pymodbus.framer import FRAMER_NAME_TO_CLASS, FramerBase, FramerType from pymodbus.logging import Log from pymodbus.pdu import DecodePDU, ModbusPDU -from pymodbus.transaction import SyncModbusTransactionManager +from pymodbus.transaction import TransactionManager from pymodbus.transport import CommParams from pymodbus.utilities import ModbusTransactionState @@ -27,15 +27,14 @@ def __init__( framer: FramerType, retries: int, on_connect_callback: Callable[[bool], None] | None, - comm_params: CommParams | None = None, + comm_params: CommParams, ) -> None: """Initialize a client instance. :meta private: """ ModbusClientMixin.__init__(self) # type: ignore[arg-type] - if comm_params: - self.comm_params = comm_params + self.comm_params = comm_params self.ctx = ModbusClientProtocol( (FRAMER_NAME_TO_CLASS[framer])(DecodePDU(False)), self.comm_params, @@ -116,23 +115,25 @@ def __init__( self, framer: FramerType, retries: int, - comm_params: CommParams | None = None, + comm_params: CommParams, ) -> None: """Initialize a client instance. :meta private: """ ModbusClientMixin.__init__(self) # type: ignore[arg-type] - if comm_params: - self.comm_params = comm_params + self.comm_params = comm_params self.retries = retries self.slaves: list[int] = [] # Common variables. self.framer: FramerBase = (FRAMER_NAME_TO_CLASS[framer])(DecodePDU(False)) - self.transaction = SyncModbusTransactionManager( + self.transaction = TransactionManager( + self.comm_params, + self.framer, + retries, + False, self, - self.retries, ) self.reconnect_delay_current = self.comm_params.reconnect_delay or 0 self.use_udp = False @@ -177,7 +178,7 @@ def execute(self, no_response_expected: bool, request: ModbusPDU) -> ModbusPDU: """ if not self.connect(): raise ConnectionException(f"Failed to connect[{self!s}]") - return self.transaction.execute(no_response_expected, request) + return self.transaction.sync_execute(no_response_expected, request) # ----------------------------------------------------------------------- # # Internal methods diff --git a/pymodbus/client/serial.py b/pymodbus/client/serial.py index c5ec55952..9fb7598c7 100644 --- a/pymodbus/client/serial.py +++ b/pymodbus/client/serial.py @@ -103,6 +103,7 @@ def __init__( # pylint: disable=too-many-arguments framer, retries, on_connect_callback, + self.comm_params, ) @@ -185,6 +186,7 @@ def __init__( # pylint: disable=too-many-arguments super().__init__( framer, retries, + self.comm_params, ) if "serial" not in sys.modules: raise RuntimeError( diff --git a/pymodbus/client/tcp.py b/pymodbus/client/tcp.py index 8da9146a1..1e1e3cb25 100644 --- a/pymodbus/client/tcp.py +++ b/pymodbus/client/tcp.py @@ -84,6 +84,7 @@ def __init__( # pylint: disable=too-many-arguments framer, retries, on_connect_callback, + self.comm_params, ) @@ -154,7 +155,7 @@ def __init__( reconnect_delay_max=reconnect_delay_max, timeout_connect=timeout, ) - super().__init__(framer, retries) + super().__init__(framer, retries, self.comm_params) self.socket = None @property diff --git a/pymodbus/client/udp.py b/pymodbus/client/udp.py index 74a094f08..4fc9dc819 100644 --- a/pymodbus/client/udp.py +++ b/pymodbus/client/udp.py @@ -85,6 +85,7 @@ def __init__( # pylint: disable=too-many-arguments framer, retries, on_connect_callback, + self.comm_params, ) self.source_address = source_address @@ -155,7 +156,7 @@ def __init__( reconnect_delay_max=reconnect_delay_max, timeout_connect=timeout, ) - super().__init__(framer, retries) + super().__init__(framer, retries, self.comm_params) self.socket = None @property diff --git a/pymodbus/transaction/__init__.py b/pymodbus/transaction/__init__.py index abd75682c..444ca29ae 100644 --- a/pymodbus/transaction/__init__.py +++ b/pymodbus/transaction/__init__.py @@ -1,12 +1,6 @@ """Transaction.""" __all__ = [ - "ModbusTransactionManager", - "SyncModbusTransactionManager", "TransactionManager", ] -from pymodbus.transaction.old_transaction import ( - ModbusTransactionManager, - SyncModbusTransactionManager, -) from pymodbus.transaction.transaction import TransactionManager diff --git a/pymodbus/transaction/old_transaction.py b/pymodbus/transaction/old_transaction.py deleted file mode 100644 index 6c2baa8a5..000000000 --- a/pymodbus/transaction/old_transaction.py +++ /dev/null @@ -1,213 +0,0 @@ -"""Collection of transaction based abstractions.""" -from __future__ import annotations - - -__all__ = [ - "ModbusTransactionManager", - "SyncModbusTransactionManager", -] - -from threading import RLock -from typing import TYPE_CHECKING - -from pymodbus.exceptions import ( - ModbusIOException, -) -from pymodbus.framer import ( - FramerSocket, -) -from pymodbus.logging import Log -from pymodbus.pdu import ModbusPDU -from pymodbus.utilities import ModbusTransactionState, hexlify_packets - - -if TYPE_CHECKING: - from pymodbus.client.base import ModbusBaseSyncClient - - -# --------------------------------------------------------------------------- # -# The Global Transaction Manager -# --------------------------------------------------------------------------- # -class ModbusTransactionManager: - """Implement a transaction for a manager. - - Results are keyed based on the supplied transaction id. - """ - - def __init__(self): - """Initialize an instance of the ModbusTransactionManager.""" - self.tid = 0 - self.transactions: dict[int, ModbusPDU] = {} - - def __iter__(self): - """Iterate over the current managed transactions. - - :returns: An iterator of the managed transactions - """ - return iter(self.transactions.keys()) - - def addTransaction(self, request: ModbusPDU): - """Add a transaction to the handler. - - This holds the request in case it needs to be resent. - After being sent, the request is removed. - - :param request: The request to hold on to - """ - tid = request.transaction_id - Log.debug("Adding transaction {}", tid) - self.transactions[tid] = request - - def getTransaction(self, tid: int): - """Return a transaction matching the referenced tid. - - If the transaction does not exist, None is returned - - :param tid: The transaction to retrieve - - """ - Log.debug("Getting transaction {}", tid) - if not tid: - if self.transactions: - ret = self.transactions.popitem()[1] - self.transactions.clear() - return ret - return None - return self.transactions.pop(tid, None) - - def delTransaction(self, tid: int): - """Remove a transaction matching the referenced tid. - - :param tid: The transaction to remove - """ - Log.debug("deleting transaction {}", tid) - self.transactions.pop(tid, None) - - def getNextTID(self) -> int: - """Retrieve the next unique transaction identifier. - - This handles incrementing the identifier after - retrieval - - :returns: The next unique transaction identifier - """ - if self.tid < 65000: - self.tid += 1 - else: - self.tid = 1 - return self.tid - - def reset(self): - """Reset the transaction identifier.""" - self.tid = 0 - self.transactions = {} - - -class SyncModbusTransactionManager(ModbusTransactionManager): - """Implement a transaction for a manager.""" - - def __init__(self, client: ModbusBaseSyncClient, retries): - """Initialize an instance of the ModbusTransactionManager.""" - super().__init__() - self.client: ModbusBaseSyncClient = client - self.retries = retries - self._transaction_lock = RLock() - self.databuffer = b'' - - def send_request(self, request: ModbusPDU) -> bool: - """Build and send request.""" - self.client.connect() - packet = self.client.framer.buildFrame(request) - Log.debug("SEND: {}", packet, ":hex") - if (size := self.client.send(packet)) != len(packet): - Log.error(f"Only sent {size} of {len(packet)} bytes") - return False - if self.client.comm_params.handle_local_echo and self.client.recv(size) != packet: - Log.error("Wrong local echo") - return False - return True - - def receive_response(self) -> ModbusPDU | None: - """Receive until PDU is correct or timeout.""" - while True: - if not (data := self.client.recv(None)): - return None - self.databuffer += data - used_len, pdu = self.client.framer.processIncomingFrame(self.databuffer) - self.databuffer = self.databuffer[used_len:] - if pdu: - return pdu - - def execute(self, no_response_expected: bool, request: ModbusPDU): # noqa: C901 - """Start the producer to send the next request to consumer.write(Frame(request)).""" - with self._transaction_lock: - Log.debug( - "Current transaction state - {}", - ModbusTransactionState.to_string(self.client.state), - ) - if isinstance(self.client.framer, FramerSocket): - request.transaction_id = self.getNextTID() - else: - request.transaction_id = 0 - Log.debug("Running transaction {}", request.transaction_id) - if _buffer := hexlify_packets( - self.databuffer - ): - Log.debug("Clearing current Frame: - {}", _buffer) - self.databuffer = b'' - - retry = self.retries + 1 - exp_txt = "" - while retry > 0: - if not self.send_request(request): - Log.debug('Changing transaction state from "SENDING" to "RETRYING"') - exp_txt = 'SEND failed' - Log.error(exp_txt + ', retrying') - self.client.state = ModbusTransactionState.RETRYING - retry -= 1 - continue - if no_response_expected: - Log.debug( - 'Changing transaction state from "SENDING" ' - 'to "TRANSACTION_COMPLETE" (no response expected)' - ) - self.client.state = ModbusTransactionState.TRANSACTION_COMPLETE - return None - Log.debug('Changing transaction state from "SENDING" to "WAITING_FOR_REPLY"') - self.client.state = ModbusTransactionState.WAITING_FOR_REPLY - if not (pdu := self.receive_response()): - Log.debug('Changing transaction state from "WAITING_FOR_REPLY" to "RETRYING"') - exp_txt = 'RECEIVE failed' - Log.error(exp_txt + ', retrying') - self.client.state = ModbusTransactionState.RETRYING - retry -= 1 - continue - - print(pdu) - break - - if not retry: - return ModbusIOException(exp_txt, request.function_code) - - if pdu: - self.addTransaction(pdu) - if not (result := self.getTransaction(request.transaction_id)): - if len(self.transactions): - result = self.getTransaction(0) - else: - last_exception = ( - "No Response received from the remote slave" - "/Unable to decode response" - ) - result = ModbusIOException( - last_exception, request.function_code - ) - self.client.close() - if hasattr(self.client, "state"): - Log.debug( - "Changing transaction state from " - '"PROCESSING REPLY" to ' - '"TRANSACTION_COMPLETE"' - ) - self.client.state = ModbusTransactionState.TRANSACTION_COMPLETE - return result diff --git a/pymodbus/transaction/transaction.py b/pymodbus/transaction/transaction.py index b8bc5dade..cd0c1460f 100644 --- a/pymodbus/transaction/transaction.py +++ b/pymodbus/transaction/transaction.py @@ -3,11 +3,12 @@ import asyncio from collections.abc import Callable +from threading import RLock from pymodbus.exceptions import ConnectionException, ModbusIOException from pymodbus.framer import FramerBase from pymodbus.logging import Log -from pymodbus.pdu import ModbusPDU +from pymodbus.pdu import ExceptionResponse, ModbusPDU from pymodbus.transport import CommParams, ModbusProtocol @@ -39,9 +40,10 @@ def __init__( framer: FramerBase, retries: int, is_server: bool, + sync_client = None, ) -> None: """Initialize an instance of the ModbusTransactionManager.""" - super().__init__(params, is_server) + super().__init__(params, is_server, is_sync=bool(sync_client)) self.framer = framer self.retries = retries self.next_tid: int = 0 @@ -51,14 +53,70 @@ def __init__( self.trace_send_pdu: Callable[[ModbusPDU | None], ModbusPDU] | None = None self.accept_no_response_limit = retries + 3 self.count_no_responses = 0 - self._lock = asyncio.Lock() + if sync_client: + self.sync_client = sync_client + self._sync_lock = RLock() + else: + self._lock = asyncio.Lock() self.response_future: asyncio.Future = asyncio.Future() - async def execute(self, no_response_expected: bool, request) -> ModbusPDU | None: - """Execute requests asynchronously.""" + def sync_get_response(self) -> ModbusPDU: + """Receive until PDU is correct or timeout.""" + databuffer = b'' + while True: + if not (data := self.sync_client.recv(None)): + raise asyncio.exceptions.TimeoutError() + databuffer += data + used_len, pdu = self.framer.processIncomingFrame(databuffer) + databuffer = databuffer[used_len:] + if pdu: + return pdu + + def sync_execute(self, no_response_expected: bool, request: ModbusPDU) -> ModbusPDU: + """Execute requests asynchronously. + + REMARK: this method is identical to execute, apart from the lock and sync_receive. + any changes in either method MUST be mirrored !!! + """ if not self.transport: Log.warning("Not connected, trying to connect!") - if not self.transport and not await self.connect(): + if not self.sync_client.connect(): + raise ConnectionException("Client cannot connect (automatic retry continuing) !!") + with self._sync_lock: + request.transaction_id = self.getNextTID() + if self.trace_send_pdu: + request = self.trace_send_pdu(request) # pylint: disable=not-callable + packet = self.framer.buildFrame(request) + count_retries = 0 + while count_retries <= self.retries: + if self.trace_send_packet: + packet = self.trace_send_packet(packet) # pylint: disable=not-callable + self.sync_client.send(packet) + if no_response_expected: + return ExceptionResponse(0xff) + try: + return self.sync_get_response() + except asyncio.exceptions.TimeoutError: + count_retries += 1 + if self.count_no_responses >= self.accept_no_response_limit: + self.connection_lost(asyncio.TimeoutError("Server not responding")) + raise ModbusIOException( + f"ERROR: No response received of the last {self.accept_no_response_limit} request, CLOSING CONNECTION." + ) + self.count_no_responses += 1 + txt = f"No response received after {self.retries} retries, continue with next request" + Log.error(txt) + raise ModbusIOException(txt) + + async def execute(self, no_response_expected: bool, request: ModbusPDU) -> ModbusPDU | None: + """Execute requests asynchronously. + + REMARK: this method is identical to sync_execute, apart from the lock and try/except. + any changes in either method MUST be mirrored !!! + """ + if not self.transport: + Log.warning("Not connected, trying to connect!") + if not await self.connect(): raise ConnectionException("Client cannot connect (automatic retry continuing) !!") async with self._lock: request.transaction_id = self.getNextTID() @@ -101,14 +159,12 @@ def callback_connected(self) -> None: def callback_disconnected(self, exc: Exception | None) -> None: """Call when connection is lost.""" - if self.trace_recv_packet: - self.trace_recv_packet(None) # pylint: disable=not-callable - if self.trace_recv_pdu: - self.trace_recv_pdu(None) # pylint: disable=not-callable - if self.trace_send_packet: - self.trace_send_packet(None) # pylint: disable=not-callable - if self.trace_send_pdu: - self.trace_send_pdu(None) # pylint: disable=not-callable + for call in (self.trace_recv_packet, + self.trace_recv_pdu, + self.trace_send_packet, + self.trace_send_pdu): + if call: + call(None) # pylint: disable=not-callable def callback_data(self, data: bytes, addr: tuple | None = None) -> int: """Handle received data.""" diff --git a/pymodbus/transport/transport.py b/pymodbus/transport/transport.py index 6ccc84eee..36750a65a 100644 --- a/pymodbus/transport/transport.py +++ b/pymodbus/transport/transport.py @@ -138,18 +138,19 @@ def __init__( self, params: CommParams, is_server: bool, + is_sync: bool = False ) -> None: """Initialize a transport instance. :param params: parameter dataclass :param is_server: true if object act as a server (listen/connect) + :param is_sync: true if used with sync client """ self.comm_params = params.copy() self.is_server = is_server self.is_closing = False self.transport: asyncio.BaseTransport = None # type: ignore[assignment] - self.loop: asyncio.AbstractEventLoop = asyncio.get_running_loop() self.recv_buffer: bytes = b"" self.call_create: Callable[[], Coroutine[Any, Any, Any]] = None # type: ignore[assignment] self.reconnect_task: asyncio.Task | None = None @@ -159,6 +160,10 @@ def __init__( self.unique_id: str = str(id(self)) self.reconnect_delay_current = 0.0 self.sent_buffer: bytes = b"" + self.loop: asyncio.AbstractEventLoop + if is_sync: + return + self.loop = asyncio.get_running_loop() if self.is_server: if self.comm_params.source_address is not None: host = self.comm_params.source_address[0] diff --git a/test/sub_client/test_client.py b/test/sub_client/test_client.py index 38f227582..92ade6844 100755 --- a/test/sub_client/test_client.py +++ b/test/sub_client/test_client.py @@ -246,7 +246,7 @@ async def test_client_modbusbaseclient(): FramerType.ASCII, 3, None, - comm_params=CommParams( + CommParams( host="localhost", port=BASE_PORT + 1, comm_type=CommType.TCP, @@ -285,7 +285,7 @@ async def test_client_base_async(): FramerType.ASCII, 3, None, - comm_params=CommParams( + CommParams( host="localhost", port=BASE_PORT + 2, comm_type=CommType.TCP, diff --git a/test/transaction/test_old_transaction.py b/test/transaction/test_old_transaction.py deleted file mode 100755 index a7fc88d33..000000000 --- a/test/transaction/test_old_transaction.py +++ /dev/null @@ -1,123 +0,0 @@ -"""Test transaction.""" -from unittest import mock - -from pymodbus.client import ModbusTcpClient -from pymodbus.exceptions import ( - ModbusIOException, -) -from pymodbus.framer import ( - FramerAscii, - FramerRTU, - FramerSocket, - FramerTLS, -) -from pymodbus.pdu import DecodePDU, ModbusPDU -from pymodbus.transaction import ( - ModbusTransactionManager, - SyncModbusTransactionManager, -) - - -TEST_MESSAGE = b"\x7b\x01\x03\x00\x00\x00\x05\x85\xC9\x7d" - - -class TestOldTransaction: # pylint: disable=too-many-public-methods - """Unittest for the pymodbus.transaction module.""" - - client = None - decoder = None - _tcp = None - _tls = None - _rtu = None - _ascii = None - _manager = None - _tm = None - - # ----------------------------------------------------------------------- # - # Test Construction - # ----------------------------------------------------------------------- # - def setup_method(self): - """Set up the test environment.""" - self.decoder = DecodePDU(True) - self._tcp = FramerSocket(self.decoder) - self._tls = FramerTLS(self.decoder) - self._rtu = FramerRTU(self.decoder) - self._ascii = FramerAscii(self.decoder) - client = mock.MagicMock() - client.framer = self._rtu - self._manager = SyncModbusTransactionManager(client, 3) - - # ----------------------------------------------------------------------- # - # Modbus transaction manager - # ----------------------------------------------------------------------- # - - @mock.patch.object(ModbusTransactionManager, "getTransaction") - def test_execute(self, mock_get_transaction): - """Test execute.""" - client = ModbusTcpClient("localhost") - client.recv = mock.Mock() - client.framer = self._ascii - client.framer._buffer = b"deadbeef" # pylint: disable=protected-access - client.framer.processIncomingFrame = mock.MagicMock() - client.framer.processIncomingFrame.return_value = 0, None - client.framer.buildFrame = mock.MagicMock() - client.framer.buildFrame.return_value = b"deadbeef" - client.send = mock.MagicMock() - client.send.return_value = len(b"deadbeef") - request = mock.MagicMock() - request.get_response_pdu_size.return_value = 10 - request.slave_id = 1 - request.function_code = 222 - trans = SyncModbusTransactionManager(client, 3) - assert trans.retries == 3 - - client.recv.side_effect=iter([b"abcdef", None]) - mock_get_transaction.return_value = b"response" - trans.retries = 0 - response = trans.execute(False, request) - assert isinstance(response, ModbusIOException) - # No response - client.recv.side_effect=iter([b"abcdef", None]) - trans.transactions = {} - mock_get_transaction.return_value = None - response = trans.execute(False, request) - assert isinstance(response, ModbusIOException) - - # No response with retries - client.recv.side_effect=iter([b"", b"abcdef"]) - response = trans.execute(False, request) - assert isinstance(response, ModbusIOException) - - # wrong handle_local_echo - client.recv.side_effect=iter([b"abcdef", b"deadbe", b"123456"]) - client.comm_params.handle_local_echo = True - assert trans.execute(False, request).message == "[Input/Output] SEND failed" - client.comm_params.handle_local_echo = False - - # retry on invalid response - client.recv.side_effect=iter([b"", b"abcdef", b"deadbe", b"123456"]) - response = trans.execute(False, request) - assert isinstance(response, ModbusIOException) - - def test_transaction_manager_tid(self): - """Test the transaction manager TID.""" - for tid in range(1, self._manager.getNextTID() + 10): - assert tid + 1 == self._manager.getNextTID() - self._manager.reset() - assert self._manager.getNextTID() == 1 - - def test_get_transaction_manager_transaction(self): - """Test the getting a transaction from the transaction manager.""" - self._manager.reset() - handle = ModbusPDU(transaction_id=self._manager.getNextTID(), slave_id=0) - self._manager.addTransaction(handle) - result = self._manager.getTransaction(handle.transaction_id) - assert handle is result - - def test_delete_transaction_manager_transaction(self): - """Test deleting a transaction from the dict transaction manager.""" - self._manager.reset() - handle = ModbusPDU(transaction_id=self._manager.getNextTID(), slave_id=0) - self._manager.addTransaction(handle) - self._manager.delTransaction(handle.transaction_id) - assert not self._manager.getTransaction(handle.transaction_id) diff --git a/test/transaction/test_transaction.py b/test/transaction/test_transaction.py index 4db68e357..78fea0287 100755 --- a/test/transaction/test_transaction.py +++ b/test/transaction/test_transaction.py @@ -4,8 +4,9 @@ import pytest +from pymodbus.client import ModbusBaseSyncClient from pymodbus.exceptions import ConnectionException, ModbusIOException -from pymodbus.framer import FramerRTU, FramerSocket +from pymodbus.framer import FramerRTU, FramerSocket, FramerType from pymodbus.pdu import DecodePDU from pymodbus.pdu.bit_message import ReadCoilsRequest, ReadCoilsResponse from pymodbus.transaction import TransactionManager @@ -74,7 +75,7 @@ async def test_transaction_data(self, use_clc, test): transact.trace_recv_pdu.assert_called_once_with(pdu) assert transact.response_future.result() == pdu - @pytest.mark.parametrize("scenario", range(7)) + @pytest.mark.parametrize("scenario", range(6)) async def test_transaction_execute(self, use_clc, scenario): """Test tracers in disconnect.""" transact = TransactionManager(use_clc, FramerRTU(DecodePDU(False)), 5, False) @@ -87,27 +88,27 @@ async def test_transaction_execute(self, use_clc, scenario): transact.transport = None with pytest.raises(ConnectionException): await transact.execute(False, request) - elif scenario == 2: # transport not ok and connect, no trace + elif scenario == 1: # transport not ok and connect, no trace transact.transport = None transact.connect = mock.AsyncMock(return_value=1) await transact.execute(True, request) - elif scenario == 3: # transport ok, trace and send + elif scenario == 2: # transport ok, trace and send transact.trace_send_pdu = mock.Mock(return_value=request) transact.trace_send_packet = mock.Mock(return_value=b'123') await transact.execute(True, request) transact.trace_send_pdu.assert_called_once_with(request) transact.trace_send_packet.assert_called_once_with(b'\x00\x01\x00u\x00\x05\xec\x02') - elif scenario == 4: # wait receive,timeout, no_responses + elif scenario == 3: # wait receive,timeout, no_responses transact.comm_params.timeout_connect = 0.1 transact.count_no_responses = 10 transact.connection_lost = mock.Mock() with pytest.raises(ModbusIOException): await transact.execute(False, request) - elif scenario == 5: # wait receive,timeout, no_responses pass + elif scenario == 4: # wait receive,timeout, no_responses pass transact.comm_params.timeout_connect = 0.1 transact.connection_lost = mock.Mock() assert not await transact.execute(False, request) - elif scenario == 6: # response + else: # if scenario == 5: # response transact.comm_params.timeout_connect = 0.2 transact.response_future.set_result(response) resp = asyncio.create_task(transact.execute(False, request)) @@ -133,17 +134,119 @@ async def test_client_protocol_execute_outside(self, use_clc, no_resp): transact = TransactionManager(use_clc, FramerSocket(DecodePDU(False)), 5, False) transact.send = mock.Mock() request = ReadCoilsRequest(address=117, count=5) - response = ReadCoilsResponse(bits=[True, False, True, True, False]) transact.retries = 0 transact.connection_made(mock.AsyncMock()) resp = asyncio.create_task(transact.execute(no_resp, request)) await asyncio.sleep(0.2) data = b"\x00\x00\x12\x34\x00\x06\xff\x01\x01\x02\x00\x04" transact.data_received(data) - response = await resp + result = await resp if no_resp: - assert not response + assert not result else: - assert not response.isError() + assert not result.isError() + assert isinstance(result, ReadCoilsResponse) + + +@pytest.mark.parametrize("use_port", [5098]) +class TestSyncTransaction: + """Test the pymodbus.transaction module.""" + + def test_sync_transaction_instance(self, use_clc): + """Test instantiate class.""" + client = ModbusBaseSyncClient(FramerType.SOCKET, 5, use_clc) + TransactionManager(use_clc, FramerRTU(DecodePDU(False)), 5, False, sync_client=client) + TransactionManager(use_clc, FramerRTU(DecodePDU(True)), 5, True, sync_client=client) + + + @pytest.mark.parametrize("scenario", range(6)) + async def test_sync_transaction_execute(self, use_clc, scenario): + """Test tracers in disconnect.""" + client = ModbusBaseSyncClient(FramerType.SOCKET, 5, use_clc) + transact = TransactionManager(use_clc, FramerRTU(DecodePDU(False)), 5, False, sync_client=client) + transact.send = mock.Mock() + transact.sync_client.connect = mock.Mock(return_value=True) + request = ReadCoilsRequest(address=117, count=5) + response = ReadCoilsResponse(bits=[True, False, True, True, False, False, False, False]) + transact.retries = 0 + if scenario == 0: # transport not ok and no connect + transact.transport = None + transact.sync_client.connect = mock.Mock(return_value=False) + with pytest.raises(ConnectionException): + transact.sync_execute(False, request) + elif scenario == 1: # transport not ok and connect, no trace + transact.transport = None + transact.sync_client.connect = mock.Mock(return_value=True) + transact.sync_execute(True, request) + elif scenario == 2: # transport ok, trace and send + transact.trace_send_pdu = mock.Mock(return_value=request) + transact.trace_send_packet = mock.Mock(return_value=b'123') + transact.sync_execute(True, request) + transact.trace_send_pdu.assert_called_once_with(request) + transact.trace_send_packet.assert_called_once_with(b'\x00\x01\x00u\x00\x05\xec\x02') + elif scenario == 3: # wait receive,timeout, no_responses + transact.comm_params.timeout_connect = 0.1 + transact.count_no_responses = 10 + with pytest.raises(ModbusIOException): + transact.sync_execute(False, request) + elif scenario == 4: # wait receive,timeout, no_responses pass + transact.comm_params.timeout_connect = 0.1 + with pytest.raises(ModbusIOException): + transact.sync_execute(False, request) + else: # if scenario == 5 # response + transact.transport = 1 + resp_bytes = transact.framer.buildFrame(response) + transact.sync_client.recv = mock.Mock(return_value=resp_bytes) + transact.sync_client.send = mock.Mock() + transact.comm_params.timeout_connect = 0.2 + resp = transact.sync_execute(False, request) + assert response.bits == resp.bits + + def test_sync_transaction_receiver(self, use_clc): + """Test tracers in disconnect.""" + client = ModbusBaseSyncClient(FramerType.SOCKET, 5, use_clc) + transact = TransactionManager(use_clc, FramerRTU(DecodePDU(False)), 5, False, sync_client=client) + transact.sync_client.send = mock.Mock() + request = ReadCoilsRequest(address=117, count=5) + response = ReadCoilsResponse(bits=[True, False, True, True, False, False, False, False]) + transact.retries = 0 + transact.transport = 1 + resp_bytes = transact.framer.buildFrame(response) + transact.sync_client.recv = mock.Mock(return_value=resp_bytes) + transact.sync_client.send = mock.Mock() + transact.comm_params.timeout_connect = 0.2 + resp = transact.sync_execute(False, request) + assert response.bits == resp.bits + + @pytest.mark.parametrize("no_resp", [False, True]) + def test_sync_client_protocol_execute_outside(self, use_clc, no_resp): + """Test the transaction execute method.""" + client = ModbusBaseSyncClient(FramerType.SOCKET, 5, use_clc) + transact = TransactionManager(use_clc, FramerRTU(DecodePDU(False)), 5, False, sync_client=client) + request = ReadCoilsRequest(address=117, count=5) + response = ReadCoilsResponse(bits=[True, False, True, True, False, False, False, False]) + transact.retries = 0 + transact.transport = 1 + resp_bytes = transact.framer.buildFrame(response) + transact.sync_client.recv = mock.Mock(return_value=resp_bytes) + transact.sync_client.send = mock.Mock() + result = transact.sync_execute(no_resp, request) + if no_resp: + assert result.isError() + else: + assert not result.isError() assert isinstance(response, ReadCoilsResponse) + def test_sync_client_protocol_execute_no_pdu(self, use_clc): + """Test the transaction execute method.""" + client = ModbusBaseSyncClient(FramerType.SOCKET, 5, use_clc) + transact = TransactionManager(use_clc, FramerRTU(DecodePDU(False)), 5, False, sync_client=client) + request = ReadCoilsRequest(address=117, count=5) + response = ReadCoilsResponse(bits=[True, False, True, True, False, False, False, False]) + transact.retries = 0 + transact.transport = 1 + resp_bytes = transact.framer.buildFrame(response)[:-1] + transact.sync_client.recv = mock.Mock(side_effect=[resp_bytes, b'']) + transact.sync_client.send = mock.Mock() + with pytest.raises(ModbusIOException): + transact.sync_execute(False, request)