From c93eaf4b636d1d6813b65e593731c23263ef757f Mon Sep 17 00:00:00 2001 From: James Hilliard Date: Sun, 21 Jul 2024 17:19:55 -0600 Subject: [PATCH] Use mock.patch.object to avoid protected access errors. --- test/framers/test_tbc_transaction.py | 58 +++++++++++++--------------- 1 file changed, 27 insertions(+), 31 deletions(-) diff --git a/test/framers/test_tbc_transaction.py b/test/framers/test_tbc_transaction.py index cd78d1e75..f2d956311 100755 --- a/test/framers/test_tbc_transaction.py +++ b/test/framers/test_tbc_transaction.py @@ -12,6 +12,7 @@ ModbusRtuFramer, ModbusSocketFramer, ModbusTlsFramer, + ModbusTransactionManager, SyncModbusTransactionManager, ) @@ -90,7 +91,9 @@ def test_calculate_exception_length(self): ) @mock.patch("pymodbus.transaction.time") - def test_execute(self, mock_time): + @mock.patch.object(SyncModbusTransactionManager, "_recv") + @mock.patch.object(ModbusTransactionManager, "getTransaction") + def test_execute(self, mock_get_transaction, mock_recv, mock_time): """Test execute.""" mock_time.time.side_effect = count() @@ -114,36 +117,34 @@ def test_execute(self, mock_time): request.slave_id = 1 request.function_code = 222 trans = SyncModbusTransactionManager(client, 0.3, False, False, 3) - trans._recv = mock.MagicMock( # pylint: disable=protected-access + mock_recv.reset_mock( return_value=b"abcdef" ) assert trans.retries == 3 assert not trans.retry_on_empty - trans.getTransaction = mock.MagicMock() - trans.getTransaction.return_value = "response" + mock_get_transaction.return_value = b"response" response = trans.execute(request) - assert response == "response" + assert response == b"response" # No response - trans._recv = mock.MagicMock( # pylint: disable=protected-access + mock_recv.reset_mock( return_value=b"abcdef" ) trans.transactions = {} - trans.getTransaction = mock.MagicMock() - trans.getTransaction.return_value = None + mock_get_transaction.return_value = None response = trans.execute(request) assert isinstance(response, ModbusIOException) # No response with retries trans.retry_on_empty = True - trans._recv = mock.MagicMock( # pylint: disable=protected-access + mock_recv.reset_mock( side_effect=iter([b"", b"abcdef"]) ) response = trans.execute(request) assert isinstance(response, ModbusIOException) # wrong handle_local_echo - trans._recv = mock.MagicMock( # pylint: disable=protected-access + mock_recv.reset_mock( side_effect=iter([b"abcdef", b"deadbe", b"123456"]) ) client.comm_params.handle_local_echo = True @@ -154,14 +155,14 @@ def test_execute(self, mock_time): # retry on invalid response trans.retry_on_invalid = True - trans._recv = mock.MagicMock( # pylint: disable=protected-access + mock_recv.reset_mock( side_effect=iter([b"", b"abcdef", b"deadbe", b"123456"]) ) response = trans.execute(request) assert isinstance(response, ModbusIOException) # Unable to decode response - trans._recv = mock.MagicMock( # pylint: disable=protected-access + mock_recv.reset_mock( side_effect=ModbusIOException() ) client.framer.processIncomingPacket.side_effect = mock.MagicMock( @@ -178,12 +179,11 @@ def test_execute(self, mock_time): # Broadcast w/ Local echo client.comm_params.handle_local_echo = True client.params.broadcast_enable = True - recv = mock.MagicMock(return_value=b"deadbeef") - trans._recv = recv # pylint: disable=protected-access + mock_recv.reset_mock(return_value=b"deadbeef") request.slave_id = 0 response = trans.execute(request) assert response == b"Broadcast write sent - no response expected" - recv.assert_called_once_with(8, False) + mock_recv.assert_called_once_with(8, False) client.comm_params.handle_local_echo = False def test_transaction_manager_tid(self): @@ -353,19 +353,18 @@ def callback(data): # for name in ("transaction_id", "protocol_id", "slave_id"): # assert getattr(expected, name) == getattr(actual, name) - def test_tcp_framer_packet(self): + @mock.patch.object(ModbusRequest, "encode") + def test_tcp_framer_packet(self, mock_encode): """Test a tcp frame packet build.""" - old_encode = ModbusRequest.encode - ModbusRequest.encode = lambda self: b"" message = ModbusRequest(0, 0, 0, False) message.transaction_id = 0x0001 message.protocol_id = 0x0000 message.slave_id = 0xFF message.function_code = 0x01 expected = b"\x00\x01\x00\x00\x00\x02\xff\x01" + mock_encode.return_value = b"" actual = self._tcp.buildPacket(message) assert expected == actual - ModbusRequest.encode = old_encode # ----------------------------------------------------------------------- # # TLS tests @@ -508,16 +507,15 @@ def callback(data): self._tcp.processIncomingPacket(msg, callback, [0, 1]) assert result - def test_framer_tls_framer_packet(self): + @mock.patch.object(ModbusRequest, "encode") + def test_framer_tls_framer_packet(self, mock_encode): """Test a tls frame packet build.""" - old_encode = ModbusRequest.encode - ModbusRequest.encode = lambda self: b"" message = ModbusRequest(0, 0, 0, False) message.function_code = 0x01 expected = b"\x01" + mock_encode.return_value = b"" actual = self._tls.buildPacket(message) assert expected == actual - ModbusRequest.encode = old_encode # ----------------------------------------------------------------------- # # RTU tests @@ -585,17 +583,16 @@ def callback(data): assert int(msg[0]) == header_dict["uid"] assert msg[-2:] == header_dict["crc"] - def test_rtu_framer_packet(self): + @mock.patch.object(ModbusRequest, "encode") + def test_rtu_framer_packet(self, mock_encode): """Test a rtu frame packet build.""" - old_encode = ModbusRequest.encode - ModbusRequest.encode = lambda self: b"" message = ModbusRequest(0, 0, 0, False) message.slave_id = 0xFF message.function_code = 0x01 expected = b"\xff\x01\x81\x80" # only header + CRC - no data + mock_encode.return_value = b"" actual = self._rtu.buildPacket(message) assert expected == actual - ModbusRequest.encode = old_encode def test_rtu_decode_exception(self): """Test that the RTU framer can decode errors.""" @@ -694,17 +691,16 @@ def test_ascii_framer_populate(self): self._ascii.populateResult(request) assert not request.slave_id - def test_ascii_framer_packet(self): + @mock.patch.object(ModbusRequest, "encode") + def test_ascii_framer_packet(self, mock_encode): """Test a ascii frame packet build.""" - old_encode = ModbusRequest.encode - ModbusRequest.encode = lambda self: b"" message = ModbusRequest(0, 0, 0, False) message.slave_id = 0xFF message.function_code = 0x01 expected = b":FF0100\r\n" + mock_encode.return_value = b"" actual = self._ascii.buildPacket(message) assert expected == actual - ModbusRequest.encode = old_encode def test_ascii_process_incoming_packets(self): """Test ascii process incoming packet."""