Refactor system tests with helpers

Author: tiagocoutinho

tests/system/responses/test_exception_async_responses.py

@@ -2,13 +2,19 @@
 import struct
 from functools import partial
 
-from ..validators import validate_response_mbap
+from ..validators import validate_response_mbap, validate_response_error
 from umodbus.client import tcp
 
 
 pytestmark = pytest.mark.asyncio
 
 
+async def req_rep(adu, reader, writer):
+    writer.write(adu)
+    await writer.drain()
+    return await reader.read(1024)
+
+
 @pytest.mark.parametrize('function_code, quantity', [
     (1, 0),
     (2, 0),
@@ -20,20 +26,17 @@
     (4, 0x007D + 1),
 ])
 async def test_request_returning_invalid_data_value_error(async_tcp_streams, mbap, function_code,
-                                                    quantity):
+                                                          quantity):
     """ Validate response PDU of request returning exception response with
     error code 3.
     """
-    function_code, starting_address, quantity = (function_code, 0, quantity)
+    starting_address = 0
     adu = mbap + struct.pack('>BHH', function_code, starting_address, quantity)
 
-    reader, writer = async_tcp_streams
-    writer.write(adu)
-    await writer.drain()
-    resp = await reader.read(1024)
+    resp = await req_rep(adu, *async_tcp_streams)
 
     validate_response_mbap(mbap, resp)
-    assert struct.unpack('>BB', resp[-2:]) == (0x80 + function_code, 3)
+    validate_response_error(resp, function_code, 3)
 
 
 @pytest.mark.parametrize('function', [
@@ -55,13 +58,10 @@ async def test_request_returning_invalid_data_address_error(async_tcp_streams, f
     mbap = adu[:7]
     function_code = struct.unpack('>B', adu[7:8])[0]
 
-    reader, writer = async_tcp_streams
-    writer.write(adu)
-    await writer.drain()
-    resp = await reader.read(1024)
+    resp = await req_rep(adu, *async_tcp_streams)
 
     validate_response_mbap(mbap, resp)
-    assert struct.unpack('>BB', resp[-2:]) == (0x80 + function_code, 2)
+    validate_response_error(resp, function_code, 2)
 
 
 @pytest.mark.parametrize('function', [
@@ -83,10 +83,7 @@ async def test_request_returning_server_device_failure_error(async_tcp_streams,
     mbap = adu[:7]
     function_code = struct.unpack('>B', adu[7:8])[0]
 
-    reader, writer = async_tcp_streams
-    writer.write(adu)
-    await writer.drain()
-    resp = await reader.read(1024)
+    resp = await req_rep(adu, *async_tcp_streams)
 
     validate_response_mbap(mbap, resp)
-    assert struct.unpack('>BB', resp[-2:]) == (0x80 + function_code, 4)
+    validate_response_error(resp, function_code, 4)

tests/system/responses/test_exception_async_rtu_responses.py

@@ -2,7 +2,7 @@
 import struct
 from functools import partial
 
-from ..validators import validate_response_mbap
+from ..validators import validate_response_error
 
 from umodbus.client.serial import rtu
 from umodbus.client.serial.redundancy_check import (get_crc, validate_crc,
@@ -12,6 +12,12 @@
 pytestmark = pytest.mark.asyncio
 
 
+async def req_rep(adu, reader, writer, serial_port):
+    writer.write(adu)
+    await writer.drain()
+    return await reader.read(serial_port.in_waiting)
+
+
 @pytest.mark.parametrize('function_code, quantity', [
     (1, 0),
     (2, 0),
@@ -33,12 +39,10 @@ async def test_request_returning_invalid_data_value_error(
                               starting_address, quantity))
 
     reader, writer = async_serial_streams
-    writer.write(adu)
-    await writer.drain()
-    resp = await reader.read(rtu_server.serial_port.in_waiting)
+    resp = await req_rep(adu, reader, writer, rtu_server.serial_port)
 
     validate_crc(resp)
-    assert struct.unpack('>BB', resp[1:-2]) == (0x80 + function_code, 3)
+    validate_response_error(resp[:-2], function_code, 3)
 
 
 @pytest.mark.parametrize('function', [
@@ -60,12 +64,10 @@ async def test_request_returning_invalid_data_address_error(rtu_server, async_se
     function_code = struct.unpack('>B', adu[1:2])[0]
 
     reader, writer = async_serial_streams
-    writer.write(adu)
-    await writer.drain()
-    resp = await reader.read(rtu_server.serial_port.in_waiting)
+    resp = await req_rep(adu, reader, writer, rtu_server.serial_port)
 
     validate_crc(resp)
-    assert struct.unpack('>BB', resp[1:-2]) == (0x80 + function_code, 2)
+    validate_response_error(resp[:-2], function_code, 2)
 
 
 @pytest.mark.parametrize('function', [
@@ -87,9 +89,8 @@ async def test_request_returning_server_device_failure_error(rtu_server, async_s
     function_code = struct.unpack('>B', adu[1:2])[0]
 
     reader, writer = async_serial_streams
-    writer.write(adu)
-    await writer.drain()
-    resp = await reader.read(rtu_server.serial_port.in_waiting)
+    resp = await req_rep(adu, reader, writer, rtu_server.serial_port)
 
     validate_crc(resp)
-    assert struct.unpack('>BB', resp[1:-2]) == (0x80 + function_code, 4)
+    validate_response_error(resp[:-2], function_code, 4)
+

tests/system/responses/test_exception_responses.py

@@ -2,7 +2,7 @@
 import struct
 from functools import partial
 
-from ..validators import validate_response_mbap
+from ..validators import validate_response_mbap, validate_response_error
 from umodbus.client import tcp
 
 
@@ -28,7 +28,7 @@ def test_request_returning_invalid_data_value_error(sock, mbap, function_code,
     resp = sock.recv(1024)
 
     validate_response_mbap(mbap, resp)
-    assert struct.unpack('>BB', resp[-2:]) == (0x80 + function_code, 3)
+    validate_response_error(resp, function_code, 3)
 
 
 @pytest.mark.parametrize('function', [
@@ -54,7 +54,7 @@ def test_request_returning_invalid_data_address_error(sock, function):
     resp = sock.recv(1024)
 
     validate_response_mbap(mbap, resp)
-    assert struct.unpack('>BB', resp[-2:]) == (0x80 + function_code, 2)
+    validate_response_error(resp, function_code, 2)
 
 
 @pytest.mark.parametrize('function', [
@@ -80,4 +80,4 @@ def test_request_returning_server_device_failure_error(sock, function):
     resp = sock.recv(1024)
 
     validate_response_mbap(mbap, resp)
-    assert struct.unpack('>BB', resp[-2:]) == (0x80 + function_code, 4)
+    validate_response_error(resp, function_code, 4)

tests/system/responses/test_exception_rtu_responses.py

@@ -2,8 +2,7 @@
 import struct
 from functools import partial
 
-from ..validators import validate_response_mbap
-
+from ..validators import validate_response_mbap, validate_response_error
 from umodbus.client.serial import rtu
 from umodbus.client.serial.redundancy_check import (get_crc, validate_crc,
                                                     add_crc, CRCError)
@@ -45,7 +44,7 @@ def test_request_returning_invalid_data_value_error(rtu_server, function_code,
     resp = rtu_server.serial_port.read(rtu_server.serial_port.in_waiting)
 
     validate_crc(resp)
-    assert struct.unpack('>BB', resp[1:-2]) == (0x80 + function_code, 3)
+    validate_response_error(resp[:-2], function_code, 3)
 
 
 @pytest.mark.parametrize('function', [
@@ -71,7 +70,7 @@ def test_request_returning_invalid_data_address_error(rtu_server, function):
     resp = rtu_server.serial_port.read(rtu_server.serial_port.in_waiting)
 
     validate_crc(resp)
-    assert struct.unpack('>BB', resp[1:-2]) == (0x80 + function_code, 2)
+    validate_response_error(resp[:-2], function_code, 2)
 
 
 @pytest.mark.parametrize('function', [
@@ -97,4 +96,4 @@ def test_request_returning_server_device_failure_error(rtu_server, function):
     resp = rtu_server.serial_port.read(rtu_server.serial_port.in_waiting)
 
     validate_crc(resp)
-    assert struct.unpack('>BB', resp[1:-2]) == (0x80 + function_code, 4)
+    validate_response_error(resp[:-2], function_code, 4)

tests/system/responses/test_succesful_async_responses.py

@@ -27,7 +27,9 @@ async def test_response_on_single_bit_value_read_requests(async_tcp_streams, fun
     """ Validate response of a succesful Read Coils or Read Discrete Inputs
     request.
     """
-    slave_id, starting_address, quantity = (1, 0, 10)
+    slave_id = 1
+    starting_address = 0
+    quantity = 10
     req_adu = function(slave_id, starting_address, quantity)
 
     assert await tcp.async_send_message(req_adu, *async_tcp_streams) == [0, 1, 0, 1, 0, 1, 0, 1, 0,  1]
@@ -41,7 +43,9 @@ async def test_response_on_multi_bit_value_read_requests(async_tcp_streams, func
     """ Validate response of a succesful Read Holding Registers or Read
     Input Registers request.
     """
-    slave_id, starting_address, quantity = (1, 0, 10)
+    slave_id = 1
+    starting_address = 0
+    quantity = 10
     req_adu = function(slave_id, starting_address, quantity)
 
     assert await tcp.async_send_message(req_adu, *async_tcp_streams) ==\
@@ -56,7 +60,9 @@ async def test_response_single_value_write_request(async_tcp_streams, function,
     """ Validate responde of succesful Read Single Coil and Read Single
     Register request.
     """
-    slave_id, starting_address, value = (1, 0, value)
+    slave_id = 1
+    starting_address = 0
+    quantity = 10
     req_adu = function(slave_id, starting_address, value)
 
     assert await tcp.async_send_message(req_adu, *async_tcp_streams) == value
@@ -72,7 +78,9 @@ async def test_async_response_multi_value_write_request(async_tcp_streams, funct
 
     Both requests write 2 values, starting address is 0.
     """
-    slave_id, starting_address = (1, 0)
+    slave_id = 1
+    starting_address = 0
+    quantity = 10
     req_adu = function(slave_id, starting_address, values)
 
     assert await tcp.async_send_message(req_adu, *async_tcp_streams) == 2

tests/system/validators.py

@@ -32,6 +32,11 @@ def validate_response_mbap(request_mbap, response):
     validate_unit_id(request_mbap, response)
 
 
+def validate_response_error(resp, function_code, error_code):
+    assert struct.unpack('>BB', resp[-2:]) == \
+        (0x80 + function_code, error_code)
+
+
 def validate_function_code(request, response):
     """ Validate if Function code in request and response equal. """
     assert struct.unpack('>B', request[7:8])[0] == \