Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Use mock.patch.object to avoid protected access errors. #2251

Merged
merged 1 commit into from
Jul 21, 2024
Merged

Use mock.patch.object to avoid protected access errors. #2251

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
55 changes: 26 additions & 29 deletions test/test_transaction.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
ModbusRtuFramer,
ModbusSocketFramer,
ModbusTlsFramer,
ModbusTransactionManager,
SyncModbusTransactionManager,
)

Expand Down Expand Up @@ -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()

Expand All @@ -114,35 +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(return_value = b"response")
mock_get_transaction.return_value = b"response"
response = trans.execute(request)
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
Expand All @@ -153,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(
Expand All @@ -177,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):
Expand Down Expand Up @@ -339,19 +340,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
Expand Down Expand Up @@ -494,16 +494,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
Expand Down Expand Up @@ -571,17 +570,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."""
Expand Down Expand Up @@ -680,17 +678,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."""
Expand Down