Please black (test). (#899)

Author: janiversen

test/asyncio_test_helper.py

@@ -1,4 +1,3 @@
-
 """Asyncio helper."""
 import functools
 

test/test_all_messages.py

@@ -44,8 +44,10 @@ class ModbusAllMessagesTests(unittest.TestCase):
     def setUp(self):
         """Initialize the test environment and builds request/result encoding pairs."""
         arguments = {
-            "read_address": 1, "read_count": 1,
-            "write_address": 1, "write_registers": 1
+            "read_address": 1,
+            "read_count": 1,
+            "write_address": 1,
+            "write_registers": 1,
         }
         self.requests = [
             lambda unit: ReadCoilsRequest(1, 5, unit=unit),

test/test_bit_read_messages.py

@@ -15,6 +15,7 @@
 from pymodbus.pdu import ModbusExceptions
 
 from .modbus_mocks import MockContext
+
 res = [True] * 21
 res.extend([False] * 3)
 # ---------------------------------------------------------------------------#
@@ -77,7 +78,7 @@ def test_bit_read_base_requests(self):
         """Test bit read request encoding"""
         messages = {
             ReadBitsRequestBase(12, 14): b"\x00\x0c\x00\x0e",
-            ReadBitsResponseBase([1, 0, 1, 1, 0]): b"\x01\x0d"
+            ReadBitsResponseBase([1, 0, 1, 1, 0]): b"\x01\x0d",
         }
         for request, expected in iter(messages.items()):
             self.assertEqual(request.encode(), expected)
@@ -91,8 +92,7 @@ def test_bit_read_message_execute_value_errors(self):
         ]
         for request in requests:
             result = request.execute(context)
-            self.assertEqual(ModbusExceptions.IllegalValue,
-                             result.exception_code)
+            self.assertEqual(ModbusExceptions.IllegalValue, result.exception_code)
 
     def test_bit_read_message_execute_address_errors(self):
         """Test bit read request encoding"""

test/test_bit_write_messages.py

@@ -39,8 +39,8 @@ def tearDown(self):
     def test_bit_write_base_requests(self):
         """Test bit write base."""
         messages = {
-            WriteSingleCoilRequest(1, 0xabcd): b"\x00\x01\xff\x00",
-            WriteSingleCoilResponse(1, 0xabcd): b"\x00\x01\xff\x00",
+            WriteSingleCoilRequest(1, 0xABCD): b"\x00\x01\xff\x00",
+            WriteSingleCoilResponse(1, 0xABCD): b"\x00\x01\xff\x00",
             WriteMultipleCoilsRequest(1, [True] * 5): b"\x00\x01\x00\x05\x01\x1f",
             WriteMultipleCoilsResponse(1, 5): b"\x00\x01\x00\x05",
         }
@@ -49,9 +49,7 @@ def test_bit_write_base_requests(self):
 
     def test_bit_write_message_get_response_pdu(self):
         """Test bit write message."""
-        requests = {
-            WriteSingleCoilRequest(1, 0xabcd): 5
-        }
+        requests = {WriteSingleCoilRequest(1, 0xABCD): 5}
         for request, expected in iter(requests.items()):
             pdu_len = request.get_response_pdu_size()
             self.assertEqual(pdu_len, expected)
@@ -126,8 +124,8 @@ def test_write_multiple_coils_response(self):
     def test_serializing_to_string(self):
         """Test serializing to string."""
         requests = [
-            WriteSingleCoilRequest(1, 0xabcd),
-            WriteSingleCoilResponse(1, 0xabcd),
+            WriteSingleCoilRequest(1, 0xABCD),
+            WriteSingleCoilResponse(1, 0xABCD),
             WriteMultipleCoilsRequest(1, [True] * 5),
             WriteMultipleCoilsResponse(1, 5),
         ]

test/test_client_async.py

@@ -17,10 +17,14 @@
 from pymodbus.client.asynchronous.tls import AsyncModbusTLSClient
 from pymodbus.client.asynchronous.udp import AsyncModbusUDPClient
 from pymodbus.client.asynchronous.tornado import (
-    AsyncModbusSerialClient as AsyncTornadoModbusSerialClient
+    AsyncModbusSerialClient as AsyncTornadoModbusSerialClient,
+)
+from pymodbus.client.asynchronous.tornado import (
+    AsyncModbusTCPClient as AsyncTornadoModbusTcpClient,
+)
+from pymodbus.client.asynchronous.tornado import (
+    AsyncModbusUDPClient as AsyncTornadoModbusUdoClient,
 )
-from pymodbus.client.asynchronous.tornado import AsyncModbusTCPClient as AsyncTornadoModbusTcpClient
-from pymodbus.client.asynchronous.tornado import AsyncModbusUDPClient as AsyncTornadoModbusUdoClient
 from pymodbus.client.asynchronous import schedulers
 from pymodbus.factory import ClientDecoder
 from pymodbus.exceptions import ConnectionException
@@ -58,20 +62,26 @@ class TestAsynchronousClient:
     def test_tcp_twisted_client(self):  # pylint: disable=no-self-use
         """Test the TCP Twisted client."""
         with patch("twisted.internet.reactor"):
+
             def test_callback(client):  # pylint: disable=unused-argument
                 pass
 
-            AsyncModbusTCPClient(schedulers.REACTOR,
-                                 framer=ModbusSocketFramer(ClientDecoder()),
-                                 callback=test_callback,
-                                 errback=test_callback)
+            AsyncModbusTCPClient(
+                schedulers.REACTOR,
+                framer=ModbusSocketFramer(ClientDecoder()),
+                callback=test_callback,
+                errback=test_callback,
+            )
 
     @patch("pymodbus.client.asynchronous.tornado.IOLoop")
     @patch("pymodbus.client.asynchronous.tornado.IOStream")
-    def test_tcp_tornado_client(self, mock_iostream, mock_ioloop):  # pylint: disable=no-self-use,unused-argument
+    def test_tcp_tornado_client(
+        self, mock_iostream, mock_ioloop
+    ):  # pylint: disable=no-self-use,unused-argument
         """Test the TCP tornado client client initialize"""
-        protocol, future = AsyncModbusTCPClient(  # NOSOANR pylint: disable=unpacking-non-sequence
-            schedulers.IO_LOOP, framer=ModbusSocketFramer(ClientDecoder()))
+        protocol, future = AsyncModbusTCPClient(  # NOSONAR pylint: disable=unpacking-non-sequence
+            schedulers.IO_LOOP, framer=ModbusSocketFramer(ClientDecoder())
+        )
         client = future.result()
         assert isinstance(client, AsyncTornadoModbusTcpClient)  # nosec
         assert not list(client.transaction)  # nosec
@@ -94,7 +104,9 @@ def handle_failure(failure):
 
     @patch("asyncio.get_event_loop")
     @patch("asyncio.gather")
-    def test_tcp_asyncio_client(self, mock_gather, mock_loop):  # pylint: disable=no-self-use,unused-argument
+    def test_tcp_asyncio_client(
+        self, mock_gather, mock_loop
+    ):  # pylint: disable=no-self-use,unused-argument
         """Test the TCP Twisted client."""
         pytest.skip("TBD")
 
@@ -104,7 +116,9 @@ def test_tcp_asyncio_client(self, mock_gather, mock_loop):  # pylint: disable=no
 
     def test_tls_asyncio_client(self):  # pylint: disable=no-self-use
         """Test the TLS AsyncIO client."""
-        _, client = AsyncModbusTLSClient(schedulers.ASYNC_IO)  # NOSONAR pylint: disable=unpacking-non-sequence
+        _, client = AsyncModbusTLSClient(  # NOSONAR pylint: disable=unpacking-non-sequence
+            schedulers.ASYNC_IO
+        )
         assert isinstance(client, ReconnectingAsyncioModbusTlsClient)  # nosec
         assert isinstance(client.framer, ModbusTlsFramer)  # nosec
         assert isinstance(client.sslctx, ssl.SSLContext)  # nosec
@@ -119,10 +133,13 @@ def test_tls_asyncio_client(self):  # pylint: disable=no-self-use
 
     @patch("pymodbus.client.asynchronous.tornado.IOLoop")
     @patch("pymodbus.client.asynchronous.tornado.IOStream")
-    def test_udp_tornado_client(self, mock_iostream, mock_ioloop):  # pylint: disable=no-self-use,unused-argument
+    def test_udp_tornado_client(
+        self, mock_iostream, mock_ioloop
+    ):  # pylint: disable=no-self-use,unused-argument
         """Test the udp tornado client client initialize"""
         protocol, future = AsyncModbusUDPClient(  # NOSONAR pylint: disable=unpacking-non-sequence
-            schedulers.IO_LOOP, framer=ModbusSocketFramer(ClientDecoder()))
+            schedulers.IO_LOOP, framer=ModbusSocketFramer(ClientDecoder())
+        )
         client = future.result()
         assert isinstance(client, AsyncTornadoModbusUdoClient)  # nosec
         assert not list(client.transaction)  # nosec
@@ -144,12 +161,15 @@ def handle_failure(failure):
     def test_udp_twisted_client(self):  # pylint: disable=no-self-use
         """Test the udp twisted client client initialize"""
         with pytest.raises(NotImplementedError):
-            AsyncModbusUDPClient(schedulers.REACTOR,
-                                 framer=ModbusSocketFramer(ClientDecoder()))
+            AsyncModbusUDPClient(
+                schedulers.REACTOR, framer=ModbusSocketFramer(ClientDecoder())
+            )
 
     @patch("asyncio.get_event_loop")
     @patch("asyncio.gather", side_effect=mock_asyncio_gather)
-    def test_udp_asyncio_client(self, mock_gather, mock_event_loop):  # pylint: disable=no-self-use,unused-argument
+    def test_udp_asyncio_client(
+        self, mock_gather, mock_event_loop
+    ):  # pylint: disable=no-self-use,unused-argument
         """Test the udp asyncio client"""
         pytest.skip("TBD")
 
@@ -161,51 +181,78 @@ def test_udp_asyncio_client(self, mock_gather, mock_event_loop):  # pylint: disa
         sys.platform == "win32" and platform.python_implementation() == "PyPy",
         reason="Twisted serial requires pywin32 which is not compatible with PyPy",
     )
-    @pytest.mark.parametrize("method, framer", [("rtu", ModbusRtuFramer),
-                                                ("socket", ModbusSocketFramer),
-                                                ("binary", ModbusBinaryFramer),
-                                                ("ascii", ModbusAsciiFramer)])
+    @pytest.mark.parametrize(
+        "method, framer",
+        [
+            ("rtu", ModbusRtuFramer),
+            ("socket", ModbusSocketFramer),
+            ("binary", ModbusBinaryFramer),
+            ("ascii", ModbusAsciiFramer),
+        ],
+    )
     def test_serial_twisted_client(self, method, framer):  # pylint: disable=no-self-use
         """Test the serial twisted client client initialize"""
         with patch("serial.Serial"):
-            from twisted.internet.serialport import SerialPort  # pylint: disable=import-outside-toplevel
-            with maybe_manage(sys.platform == "win32", patch.object(
-                    SerialPort, "_finishPortSetup")):
+            from twisted.internet.serialport import (  # pylint: disable=import-outside-toplevel
+                SerialPort,
+            )
+
+            with maybe_manage(
+                sys.platform == "win32", patch.object(SerialPort, "_finishPortSetup")
+            ):
                 with patch("twisted.internet.reactor"):
 
                     protocol, client = AsyncModbusSerialClient(  # NOSONAR pylint: disable=unpacking-non-sequence
                         schedulers.REACTOR,
                         method=method,
                         port=pytest.SERIAL_PORT,
-                        proto_cls=ModbusSerClientProtocol
+                        proto_cls=ModbusSerClientProtocol,
                     )
 
                     assert isinstance(client, SerialPort)  # nosec
                     assert isinstance(client.protocol, ModbusSerClientProtocol)  # nosec
                     assert not list(client.protocol.transaction)  # nosec
                     assert isinstance(client.protocol.framer, framer)  # nosec
-                    assert client.protocol._connected  # nosec pylint: disable=protected-access
+                    assert (
+                        client.protocol._connected  # nosec pylint: disable=protected-access
+                    )
 
                     def handle_failure(failure):
-                        assert (isinstance(failure.exception(), ConnectionException))  # nosec
+                        assert isinstance(
+                            failure.exception(), ConnectionException  # nosec
+                        )
 
-                    response = client.protocol._buildResponse(0x00)  # pylint: disable=protected-access
+                    response = client.protocol._buildResponse(  # pylint: disable=protected-access
+                        0x00
+                    )
                     response.addCallback(handle_failure)
 
-                    assert client.protocol._connected  # nosec pylint: disable=protected-access
+                    assert (
+                        client.protocol._connected  # nosec pylint: disable=protected-access
+                    )
                     client.protocol.close()
                     protocol.stop()
-                    assert not client.protocol._connected  # nosec pylint: disable=protected-access
+                    assert (
+                        not client.protocol._connected  # nosec pylint: disable=protected-access
+                    )
 
-    @pytest.mark.parametrize("method, framer", [("rtu", ModbusRtuFramer),
-                                                ("socket", ModbusSocketFramer),
-                                                ("binary", ModbusBinaryFramer),
-                                                ("ascii", ModbusAsciiFramer)])
+    @pytest.mark.parametrize(
+        "method, framer",
+        [
+            ("rtu", ModbusRtuFramer),
+            ("socket", ModbusSocketFramer),
+            ("binary", ModbusBinaryFramer),
+            ("ascii", ModbusAsciiFramer),
+        ],
+    )
     def test_serial_tornado_client(self, method, framer):  # pylint: disable=no-self-use
         """Test the serial tornado client client initialize"""
-        with maybe_manage(sys.platform in {"darwin", "win32"}, patch.object(Serial, "open")):
+        with maybe_manage(
+            sys.platform in {"darwin", "win32"}, patch.object(Serial, "open")
+        ):
             protocol, future = AsyncModbusSerialClient(  # NOSONAR pylint: disable=unpacking-non-sequence
-                schedulers.IO_LOOP, method=method, port=pytest.SERIAL_PORT)
+                schedulers.IO_LOOP, method=method, port=pytest.SERIAL_PORT
+            )
             client = future.result()
             assert isinstance(client, AsyncTornadoModbusSerialClient)  # nosec
             assert not list(client.transaction)  # nosec
@@ -226,18 +273,35 @@ def handle_failure(failure):
 
     @patch("asyncio.get_event_loop")
     @patch("asyncio.gather", side_effect=mock_asyncio_gather)
-    @pytest.mark.parametrize("method, framer", [("rtu", ModbusRtuFramer),
-                                                ("socket", ModbusSocketFramer),
-                                                ("binary", ModbusBinaryFramer),
-                                                ("ascii", ModbusAsciiFramer)])
-    def test_serial_asyncio_client(self, mock_gather,  # pylint: disable=no-self-use,unused-argument
-                                   mock_event_loop, method, framer):  # pylint: disable=unused-argument
+    @pytest.mark.parametrize(
+        "method, framer",
+        [
+            ("rtu", ModbusRtuFramer),
+            ("socket", ModbusSocketFramer),
+            ("binary", ModbusBinaryFramer),
+            ("ascii", ModbusAsciiFramer),
+        ],
+    )
+    def test_serial_asyncio_client(  # pylint: disable=no-self-use
+        self,
+        mock_gather,  # pylint: disable=unused-argument
+        mock_event_loop,
+        method,
+        framer,
+    ):  # pylint: disable=unused-argument
         """Test that AsyncModbusSerialClient instantiates AsyncioModbusSerialClient for asyncio scheduler."""
         loop = asyncio.get_event_loop()
         loop.is_running.side_effect = lambda: False
         loop, client = AsyncModbusSerialClient(  # NOSONAR pylint: disable=unpacking-non-sequence
-            schedulers.ASYNC_IO, method=method, port=pytest.SERIAL_PORT, loop=loop,
-            baudrate=19200, parity="E", stopbits=2, bytesize=7)
+            schedulers.ASYNC_IO,
+            method=method,
+            port=pytest.SERIAL_PORT,
+            loop=loop,
+            baudrate=19200,
+            parity="E",
+            stopbits=2,
+            bytesize=7,
+        )
         assert isinstance(client, AsyncioModbusSerialClient)  # nosec
         assert isinstance(client.framer, framer)  # nosec
         assert client.port == pytest.SERIAL_PORT  # nosec
@@ -248,6 +312,7 @@ def test_serial_asyncio_client(self, mock_gather,  # pylint: disable=no-self-use
         client.stop()
         loop.stop()
 
+
 # ---------------------------------------------------------------------------#
 # Main
 # ---------------------------------------------------------------------------#