From 07bc9fc81c2e65d220707c1d579900b9509f92cb Mon Sep 17 00:00:00 2001 From: jan iversen Date: Sun, 27 Oct 2024 22:37:26 +0100 Subject: [PATCH] Clean/Finalize testing for bit functions. (#2420) --- doc/source/roadmap.rst | 2 + test/conftest.py | 33 +-- test/framer/test_framer.py | 2 +- test/pdu/test_bit.py | 182 ++++++++++++++++ test/pdu/test_bit_pdu.py | 255 ----------------------- test/pdu/test_file_message.py | 9 +- test/pdu/test_pdu.py | 6 +- test/pdu/test_register_read_messages.py | 18 +- test/pdu/test_register_write_messages.py | 14 +- 9 files changed, 225 insertions(+), 296 deletions(-) create mode 100644 test/pdu/test_bit.py delete mode 100644 test/pdu/test_bit_pdu.py diff --git a/doc/source/roadmap.rst b/doc/source/roadmap.rst index b5557e735..6b9ac8b63 100644 --- a/doc/source/roadmap.rst +++ b/doc/source/roadmap.rst @@ -20,11 +20,13 @@ The following bullet points are what the maintainers focus on: - 3.8.0, with: - on dev: - skip_encode, zero_mode parameters removed + - Simplify PDU bit classes - Simplify PDU classes - Simplify transaction manager (central control point) - Remove ModbusControlBlock - new transaction handling - transaction 100% coverage + - client 100% coverage - 4.0.0, with: - client async with sync/async API - Only one datastore, but with different API`s diff --git a/test/conftest.py b/test/conftest.py index f48c046fa..639a3d0a5 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -205,24 +205,29 @@ async def _check_system_health(): assert not NullModem.is_dirty() -class MockContext(ModbusBaseSlaveContext): - """Mock context.""" +@pytest.fixture(name="mock_context") +def define_mock_context(): + """Define context class.""" + class MockContext(ModbusBaseSlaveContext): + """Mock context.""" - def __init__(self, valid=False, default=True): - """Initialize.""" - self.valid = valid - self.default = default + def __init__(self, valid=False, default=True): + """Initialize.""" + self.valid = valid + self.default = default - def validate(self, _fc, _address, _count=0): - """Validate values.""" - return self.valid + def validate(self, _fc, _address, _count=0): + """Validate values.""" + return self.valid - def getValues(self, _fc, _address, count=0): - """Get values.""" - return [self.default] * count + def getValues(self, _fc, _address, count=0): + """Get values.""" + return [self.default] * count - def setValues(self, _fc, _address, _values): - """Set values.""" + def setValues(self, _fc, _address, _values): + """Set values.""" + + return MockContext class MockLastValuesContext(ModbusBaseSlaveContext): diff --git a/test/framer/test_framer.py b/test/framer/test_framer.py index 91164b29a..e4bb6b6e1 100644 --- a/test/framer/test_framer.py +++ b/test/framer/test_framer.py @@ -20,7 +20,7 @@ class TestFramer: """Test module.""" def test_setup(self, entry, is_server): - """Test conftest.""" + """Test setup.""" assert entry == FramerType.RTU assert not is_server set_calls() diff --git a/test/pdu/test_bit.py b/test/pdu/test_bit.py new file mode 100644 index 000000000..7b4c02a78 --- /dev/null +++ b/test/pdu/test_bit.py @@ -0,0 +1,182 @@ +"""Bit Message Test Fixture.""" + +import pymodbus.pdu.bit_message as bit_msg +from pymodbus.pdu import ModbusExceptions + + +class TestModbusBitMessage: + """Modbus bit read message tests.""" + + def test_bit_read_base_response_encoding(self): + """Test basic bit message encoding/decoding.""" + pdu = bit_msg.ReadCoilsResponse() + for i in range(1, 20): + data = [True] * i + pdu.setData(data, 0, 0) + pdu.decode(pdu.encode()) + if i % 8: + data.extend([False] * (8 - (i % 8))) + assert pdu.bits == data + + def test_bit_read_base_requests(self): + """Test bit read request encoding.""" + for pdu, args, expected in ( + (bit_msg.ReadCoilsRequest(), (12, 14, 0, 0), b"\x00\x0c\x00\x0e"), + (bit_msg.ReadCoilsResponse(), ([True, False, True, True, False], 0, 0), b"\x01\x0d"), + ): + pdu.setData(*args) + assert pdu.encode() == expected + + async def test_bit_read_update_datastore_value_errors(self, mock_context): + """Test bit read request encoding.""" + context = mock_context() + for pdu, args in ( + (bit_msg.ReadCoilsRequest(), (1, 0x800, 0, 0)), + (bit_msg.ReadDiscreteInputsRequest(), (1, 0x800, 0, 0)), + ): + pdu.setData(*args) + result = await pdu.update_datastore(context) + assert ModbusExceptions.IllegalValue == result.exception_code + + async def test_bit_read_update_datastore_address_errors(self, mock_context): + """Test bit read request encoding.""" + context = mock_context() + for pdu, args in ( + (bit_msg.ReadCoilsRequest(), (1, 5, 0, 0)), + (bit_msg.ReadDiscreteInputsRequest(), (1, 5, 0, 0)), + ): + pdu.setData(*args) + result = await pdu.update_datastore(context) + assert ModbusExceptions.IllegalAddress == result.exception_code + + async def test_bit_read_update_datastore_success(self, mock_context): + """Test bit read request encoding.""" + context = mock_context() + context.validate = lambda a, b, c: True + for pdu, args in ( + (bit_msg.ReadCoilsRequest(), (1, 5, 0, 0)), + (bit_msg.ReadDiscreteInputsRequest(), (1, 5, 0, 0)), + ): + pdu.setData(*args) + result = await pdu.update_datastore(context) + assert result.bits == [True] * 5 + + def test_bit_read_get_response_pdu(self): + """Test bit read message get response pdu.""" + for pdu, args, expected in ( + (bit_msg.ReadCoilsRequest(), (1, 5, 0, 0), 3), + (bit_msg.ReadCoilsRequest(), (1, 8, 0, 0), 3), + (bit_msg.ReadCoilsRequest(), (1, 16, 0, 0), 4), + (bit_msg.ReadDiscreteInputsRequest(), (1, 21, 0, 0), 5), + (bit_msg.ReadDiscreteInputsRequest(), (1, 24, 0, 0), 5), + (bit_msg.ReadDiscreteInputsRequest(), (1, 1900, 0, 0), 240), + ): + pdu.setData(*args) + assert pdu.get_response_pdu_size() == expected + + def test_bit_write_base_requests(self): + """Test bit write base.""" + for pdu, args, expected in ( + (bit_msg.WriteSingleCoilRequest(), (1, True, 0, 0), b"\x00\x01\xff\x00"), + (bit_msg.WriteMultipleCoilsRequest(), (1, [True] * 5, 0, 0), b"\x00\x01\x00\x05\x01\x1f"), + (bit_msg.WriteMultipleCoilsRequest(), (1, [True], 0, 0), b"\x00\x01\x00\x01\x01\x01"), + (bit_msg.WriteMultipleCoilsResponse(), (1, 5, 0, 0), b"\x00\x01\x00\x05"), + (bit_msg.WriteMultipleCoilsResponse(), (1, 1, 0, 0), b"\x00\x01\x00\x01"), + ): + pdu.setData(*args) + assert pdu.encode() == expected + + def test_write_message_get_response_pdu(self): + """Test bit write message.""" + pdu = bit_msg.WriteSingleCoilRequest() + pdu.setData(1, True, 0, 0) + pdu_len = pdu.get_response_pdu_size() + assert pdu_len == 5 + + def test_write_multiple_coils_request(self): + """Test write multiple coils.""" + request = bit_msg.WriteMultipleCoilsRequest() + for args, frame, values, expected in ( + ((1, [True] * 5, 0, 0), b"\x00\x01\x00\x05\x01\x1f", [True] * 5, 5), + ((1, [True], 0, 0), b"\x00\x01\x00\x01\x01\x01", [True], 5), + ): + request.setData(*args) + request.decode(frame) + assert request.address == 1 + assert request.values == values + assert request.get_response_pdu_size() == expected + + def test_invalid_write_multiple_coils_request(self): + """Test write invalid multiple coils.""" + request = bit_msg.WriteMultipleCoilsRequest() + request.setData(1, None, 0, 0) + assert not request.values + + def test_write_single_coil_request_encode(self): + """Test write single coil.""" + request = bit_msg.WriteSingleCoilRequest() + request.setData(1, False, 0, 0) + assert request.encode() == b"\x00\x01\x00\x00" + + async def test_write_single_coil_update_datastore(self, mock_context): + """Test write single coil.""" + context = mock_context(False, default=True) + request = bit_msg.WriteSingleCoilRequest() + request.setData(2, True, 0, 0) + result = await request.update_datastore(context) + assert result.exception_code == ModbusExceptions.IllegalAddress + + context.valid = True + result = await request.update_datastore(context) + assert result.encode() == b"\x00\x02\xff\x00" + + context = mock_context(True, default=False) + request = bit_msg.WriteSingleCoilRequest() + request.setData(2, False, 0, 0) + result = await request.update_datastore(context) + assert result.encode() == b"\x00\x02\x00\x00" + + async def test_write_multiple_coils_update_datastore(self, mock_context): + """Test write multiple coils.""" + context = mock_context(False) + # too many values + request = bit_msg.WriteMultipleCoilsRequest() + request.setData(2, [], 0, 0) + result = await request.update_datastore(context) + assert result.exception_code == ModbusExceptions.IllegalValue + + # does not validate + context.valid = False + request = bit_msg.WriteMultipleCoilsRequest() + request.setData(2, [False] * 4, 0, 0) + result = await request.update_datastore(context) + assert result.exception_code == ModbusExceptions.IllegalAddress + + # validated request + context.valid = True + result = await request.update_datastore(context) + assert result.encode() == b"\x00\x02\x00\x04" + + def test_write_multiple_coils_response(self): + """Test write multiple coils.""" + response = bit_msg.WriteMultipleCoilsResponse() + response.decode(b"\x00\x80\x00\x08") + assert response.address == 0x80 + assert response.count == 0x08 + + def test_serializing_to_string(self): + """Test serializing to string.""" + for pdu, args in ( + (bit_msg.WriteSingleCoilRequest(), (1, True, 0, 0)), + (bit_msg.WriteSingleCoilResponse(), (1, True, 0, 0)), + (bit_msg.WriteMultipleCoilsRequest(), (1, [True] * 5, 0, 0)), + (bit_msg.WriteMultipleCoilsResponse(), (1, 5, 0, 0)), + ): + pdu.setData(*args) + assert str(pdu) + + def test_pass_falsy_value_in_write_multiple_coils_request(self): + """Test pass falsy value to write multiple coils.""" + request = bit_msg.WriteMultipleCoilsRequest() + request.setData(1, [0], 0, 0) + assert request.values == [0] diff --git a/test/pdu/test_bit_pdu.py b/test/pdu/test_bit_pdu.py deleted file mode 100644 index ae5bded2d..000000000 --- a/test/pdu/test_bit_pdu.py +++ /dev/null @@ -1,255 +0,0 @@ -"""Bit Message Test Fixture. - -This fixture tests the functionality of all the -bit based request/response messages: - -* Read/Write Discretes -* Read Coils -""" - -import pymodbus.pdu.bit_message as bit_msg -from pymodbus.pdu import ModbusExceptions - -from ..conftest import MockContext - - -res = [True] * 21 -res.extend([False] * 3) - - -class TestModbusBitMessage: - """Modbus bit read message tests.""" - - def test_bit_read_base_response_encoding(self): - """Test basic bit message encoding/decoding.""" - for i in range(20): - data = [True] * i - pdu = bit_msg.ReadCoilsResponse() - pdu.setData(data, 0, 0) - result = pdu.encode() - pdu.decode(result) - assert pdu.bits[:i] == data - - def test_bit_read_base_requests(self): - """Test bit read request encoding.""" - pdu = bit_msg.ReadCoilsResponse() - pdu.setData([1, 0, 1, 1, 0], 0, 0) - messages = { - bit_msg.ReadCoilsRequest(): b"\x00\x0c\x00\x0e", - pdu: b"\x01\x0d", - } - first = True - for request, expected in iter(messages.items()): - if first: - request.setData(12, 14, 0, 0) - first = False - assert request.encode() == expected - - async def test_bit_read_update_datastore_value_errors(self): - """Test bit read request encoding.""" - context = MockContext() - pdu1 = bit_msg.ReadCoilsRequest() - pdu1.setData(1, 0x800, 0, 0) - pdu2 = bit_msg.ReadDiscreteInputsRequest() - pdu2.setData(1, 0x800, 0, 0) - requests = [ - pdu1, - pdu2, - ] - for request in requests: - result = await request.update_datastore(context) - assert ModbusExceptions.IllegalValue == result.exception_code - - async def test_bit_read_update_datastore_address_errors(self): - """Test bit read request encoding.""" - context = MockContext() - pdu1 = bit_msg.ReadCoilsRequest() - pdu1.setData(1, 5, 0, 0) - pdu2 = bit_msg.ReadDiscreteInputsRequest() - pdu2.setData(1, 5, 0, 0) - requests = [ - pdu1, - pdu2, - ] - for request in requests: - result = await request.update_datastore(context) - assert ModbusExceptions.IllegalAddress == result.exception_code - - async def test_bit_read_update_datastore_success(self): - """Test bit read request encoding.""" - context = MockContext() - context.validate = lambda a, b, c: True - pdu1 = bit_msg.ReadCoilsRequest() - pdu1.setData(1, 5, 0, 0) - pdu2 = bit_msg.ReadDiscreteInputsRequest() - pdu2.setData(1, 5, 0, 0) - requests = [ - pdu1, - pdu2, - ] - for request in requests: - result = await request.update_datastore(context) - assert result.bits == [True] * 5 - - def test_bit_read_get_response_pdu(self): - """Test bit read message get response pdu.""" - pdu1 = bit_msg.ReadCoilsRequest() - pdu1.setData(1, 5, 0, 0) - pdu2 = bit_msg.ReadCoilsRequest() - pdu2.setData(1, 8, 0, 0) - pdu3 = bit_msg.ReadCoilsRequest() - pdu3.setData(0, 16, 0, 0) - pdu4 = bit_msg.ReadDiscreteInputsRequest() - pdu4.setData(1, 21, 0, 0) - pdu5 = bit_msg.ReadDiscreteInputsRequest() - pdu5.setData(1, 24, 0, 0) - pdu6 = bit_msg.ReadDiscreteInputsRequest() - pdu6.setData(1, 1900, 0, 0) - requests = { - pdu1: 3, - pdu2: 3, - pdu3: 4, - pdu4: 5, - pdu5: 5, - pdu6: 240, - } - for request, expected in iter(requests.items()): - pdu_len = request.get_response_pdu_size() - assert pdu_len == expected - -class TestModbusBitWriteMessage: - """Modbus bit write message tests.""" - - def test_bit_write_base_requests(self): - """Test bit write base.""" - pdu1 = bit_msg.WriteSingleCoilRequest() - pdu1.setData(1, True, 0, 0) - pdu2 = bit_msg.WriteSingleCoilResponse() - pdu2.setData(1, True, 0, 0) - pdu3 = bit_msg.WriteMultipleCoilsRequest() - pdu3.setData(1, [True] * 5, 0, 0) - pdu4 = bit_msg.WriteMultipleCoilsRequest() - pdu4.setData(1, [True], 0, 0) - pdu5 = bit_msg.WriteMultipleCoilsResponse() - pdu5.setData(1, 5, 0, 0) - pdu6 = bit_msg.WriteMultipleCoilsResponse() - pdu6.setData(1, 1, 0, 0) - messages = { - pdu1: b"\x00\x01\xff\x00", - pdu2: b"\x00\x01\xff\x00", - pdu3: b"\x00\x01\x00\x05\x01\x1f", - pdu5: b"\x00\x01\x00\x05", - pdu4: b"\x00\x01\x00\x01\x01\x01", - pdu6: b"\x00\x01\x00\x01", - } - for request, expected in iter(messages.items()): - assert request.encode() == expected - - def test_write_message_get_response_pdu(self): - """Test bit write message.""" - pdu = bit_msg.WriteSingleCoilRequest() - pdu.setData(1, True, 0, 0) - requests = {pdu: 5} - for request, expected in iter(requests.items()): - pdu_len = request.get_response_pdu_size() - assert pdu_len == expected - - def test_write_multiple_coils_request(self): - """Test write multiple coils.""" - request = bit_msg.WriteMultipleCoilsRequest() - request.setData(1, [True] * 5, 0, 0) - request.decode(b"\x00\x01\x00\x05\x01\x1f") - assert request.address == 1 - assert request.values == [True] * 5 - assert request.get_response_pdu_size() == 5 - - request = bit_msg.WriteMultipleCoilsRequest() - request.setData(1, [True], 0, 0) - request.decode(b"\x00\x01\x00\x01\x01\x01") - assert request.address == 1 - assert request.values == [True] - assert request.get_response_pdu_size() == 5 - - def test_invalid_write_multiple_coils_request(self): - """Test write invalid multiple coils.""" - request = bit_msg.WriteMultipleCoilsRequest() - request.setData(1, None, 0, 0) - assert not request.values - - def test_write_single_coil_request_encode(self): - """Test write single coil.""" - request = bit_msg.WriteSingleCoilRequest() - request.setData(1, False, 0, 0) - assert request.encode() == b"\x00\x01\x00\x00" - - async def test_write_single_coil_update_datastore(self): - """Test write single coil.""" - context = MockContext(False, default=True) - request = bit_msg.WriteSingleCoilRequest() - request.setData(2, True, 0, 0) - result = await request.update_datastore(context) - assert result.exception_code == ModbusExceptions.IllegalAddress - - context.valid = True - result = await request.update_datastore(context) - assert result.encode() == b"\x00\x02\xff\x00" - - context = MockContext(True, default=False) - request = bit_msg.WriteSingleCoilRequest() - request.setData(2, False, 0, 0) - result = await request.update_datastore(context) - assert result.encode() == b"\x00\x02\x00\x00" - - async def test_write_multiple_coils_update_datastore(self): - """Test write multiple coils.""" - context = MockContext(False) - # too many values - request = bit_msg.WriteMultipleCoilsRequest() - request.setData(2, [], 0, 0) - result = await request.update_datastore(context) - assert result.exception_code == ModbusExceptions.IllegalValue - - # does not validate - context.valid = False - request = bit_msg.WriteMultipleCoilsRequest() - request.setData(2, [False] * 4, 0, 0) - result = await request.update_datastore(context) - assert result.exception_code == ModbusExceptions.IllegalAddress - - # validated request - context.valid = True - result = await request.update_datastore(context) - assert result.encode() == b"\x00\x02\x00\x04" - - def test_write_multiple_coils_response(self): - """Test write multiple coils.""" - response = bit_msg.WriteMultipleCoilsResponse() - response.decode(b"\x00\x80\x00\x08") - assert response.address == 0x80 - assert response.count == 0x08 - - def test_serializing_to_string(self): - """Test serializing to string.""" - pdu1 = bit_msg.WriteSingleCoilRequest() - pdu1.setData(1, True, 0, 0) - pdu2 = bit_msg.WriteSingleCoilResponse() - pdu2.setData(1, True, 0, 0) - pdu3 = bit_msg.WriteMultipleCoilsRequest() - pdu3.setData(1, [True] * 5, 0, 0) - pdu4 = bit_msg.WriteMultipleCoilsResponse() - pdu4.setData(1, 5, 0, 0) - requests = [ - pdu1, - pdu2, - pdu3, - pdu4, - ] - for request in requests: - result = str(request) - assert result - - def test_pass_falsy_value_in_write_multiple_coils_request(self): - """Test pass falsy value to write multiple coils.""" - request = bit_msg.WriteMultipleCoilsRequest() - request.setData(1, [0], 0, 0) - assert request.values == [0] diff --git a/test/pdu/test_file_message.py b/test/pdu/test_file_message.py index ba790261c..2886f2913 100644 --- a/test/pdu/test_file_message.py +++ b/test/pdu/test_file_message.py @@ -16,7 +16,6 @@ WriteFileRecordRequest, WriteFileRecordResponse, ) -from test.conftest import MockContext # pylint: disable=wrong-import-order TEST_MESSAGE = b"\x00\n\x00\x08\x00\x01\x00\x02\x00\x03\x00\x04" @@ -36,9 +35,9 @@ def test_read_fifo_queue_request_decode(self): handle.decode(b"\x12\x34") assert handle.address == 0x1234 - async def test_read_fifo_queue_request(self): + async def test_read_fifo_queue_request(self, mock_context): """Test basic bit message encoding/decoding.""" - context = MockContext() + context = mock_context() handle = ReadFifoQueueRequest(0x1234) result = await handle.update_datastore(context) assert isinstance(result, ReadFifoQueueResponse) @@ -51,9 +50,9 @@ async def test_read_fifo_queue_request(self): result = await handle.update_datastore(context) assert ModbusExceptions.IllegalValue == result.exception_code - async def test_read_fifo_queue_request_error(self): + async def test_read_fifo_queue_request_error(self, mock_context): """Test basic bit message encoding/decoding.""" - context = MockContext() + context = mock_context() handle = ReadFifoQueueRequest(0x1234) handle.values = [0x00] * 32 result = await handle.update_datastore(context) diff --git a/test/pdu/test_pdu.py b/test/pdu/test_pdu.py index ae8953024..8e81be38b 100644 --- a/test/pdu/test_pdu.py +++ b/test/pdu/test_pdu.py @@ -15,8 +15,6 @@ ModbusPDU, ) -from ..conftest import MockContext - class TestPdu: """Test modbus PDU.""" @@ -241,13 +239,13 @@ def test_pdu_decode(self, pdutype, args, kwargs, frame): @pytest.mark.parametrize(("pdutype", "args", "kwargs", "frame"), requests) @pytest.mark.usefixtures("frame") - async def test_pdu_datastore(self, pdutype, args, kwargs): + async def test_pdu_datastore(self, pdutype, args, kwargs, mock_context): """Test that all PDU types can be created.""" if args: pdu = pdutype() pdu.setData(*args) else: pdu = pdutype(**kwargs) - context = MockContext() + context = mock_context() context.validate = lambda a, b, c: True assert await pdu.update_datastore(context) diff --git a/test/pdu/test_register_read_messages.py b/test/pdu/test_register_read_messages.py index dbcab1f9b..5ebaa90c3 100644 --- a/test/pdu/test_register_read_messages.py +++ b/test/pdu/test_register_read_messages.py @@ -11,8 +11,6 @@ ReadWriteMultipleRegistersResponse, ) -from ..conftest import MockContext - TEST_MESSAGE = b"\x06\x00\x0a\x00\x0b\x00\x0c" @@ -106,12 +104,12 @@ async def test_register_read_requests_count_errors(self): result = await request.update_datastore(None) assert ModbusExceptions.IllegalValue == result.exception_code - async def test_register_read_requests_validate_errors(self): + async def test_register_read_requests_validate_errors(self, mock_context): """This tests that the register request messages. will break on counts that are out of range """ - context = MockContext() + context = mock_context() requests = [ ReadHoldingRegistersRequest(-1, 5), ReadInputRegistersRequest(-1, 5), @@ -122,12 +120,12 @@ async def test_register_read_requests_validate_errors(self): result = await request.update_datastore(context) assert ModbusExceptions.IllegalAddress == result.exception_code - async def test_register_read_requests_update_datastore(self): + async def test_register_read_requests_update_datastore(self, mock_context): """This tests that the register request messages. will break on counts that are out of range """ - context = MockContext(True) + context = mock_context(True) requests = [ ReadHoldingRegistersRequest(-1, 5), ReadInputRegistersRequest(-1, 5), @@ -136,18 +134,18 @@ async def test_register_read_requests_update_datastore(self): response = await request.update_datastore(context) assert request.function_code == response.function_code - async def test_read_write_multiple_registers_request(self): + async def test_read_write_multiple_registers_request(self, mock_context): """Test read/write multiple registers.""" - context = MockContext(True) + context = mock_context(True) request = ReadWriteMultipleRegistersRequest( read_address=1, read_count=10, write_address=1, write_registers=[0x00] ) response = await request.update_datastore(context) assert request.function_code == response.function_code - async def test_read_write_multiple_registers_validate(self): + async def test_read_write_multiple_registers_validate(self, mock_context): """Test read/write multiple registers.""" - context = MockContext() + context = mock_context() context.validate = lambda f, a, c: a == 1 request = ReadWriteMultipleRegistersRequest( read_address=1, read_count=10, write_address=2, write_registers=[0x00] diff --git a/test/pdu/test_register_write_messages.py b/test/pdu/test_register_write_messages.py index e8d4ef50a..430eb0dff 100644 --- a/test/pdu/test_register_write_messages.py +++ b/test/pdu/test_register_write_messages.py @@ -10,7 +10,7 @@ WriteSingleRegisterResponse, ) -from ..conftest import MockContext, MockLastValuesContext +from ..conftest import MockLastValuesContext # ---------------------------------------------------------------------------# @@ -78,9 +78,9 @@ def test_serializing_to_string(self): for request in iter(self.write.keys()): assert str(request) - async def test_write_single_register_request(self): + async def test_write_single_register_request(self, mock_context): """Test write single register request.""" - context = MockContext() + context = mock_context() request = WriteSingleRegisterRequest(0x00, 0xF0000) result = await request.update_datastore(context) assert result.exception_code == ModbusExceptions.IllegalValue @@ -93,9 +93,9 @@ async def test_write_single_register_request(self): result = await request.update_datastore(context) assert result.function_code == request.function_code - async def test_write_multiple_register_request(self): + async def test_write_multiple_register_request(self, mock_context): """Test write multiple register request.""" - context = MockContext() + context = mock_context() request = WriteMultipleRegistersRequest(0x00, [0x00] * 10) result = await request.update_datastore(context) assert result.exception_code == ModbusExceptions.IllegalAddress @@ -154,9 +154,9 @@ async def test_mask_write_register_request_update_datastore(self): assert isinstance(result, MaskWriteRegisterResponse) assert context.last_values == [0x0AF5] - async def test_mask_write_register_request_invalid_update_datastore(self): + async def test_mask_write_register_request_invalid_update_datastore(self, mock_context): """Test write register request update_datastore with invalid data.""" - context = MockContext(valid=False, default=0x0000) + context = mock_context(valid=False, default=0x0000) handle = MaskWriteRegisterRequest(0x0000, -1, 0x1010) result = await handle.update_datastore(context) assert ModbusExceptions.IllegalValue == result.exception_code