tests: use self.assert*() instead of assert. (#853)

Author: janiversen

test/test_client_sync.py

@@ -322,7 +322,7 @@ class CustomRequest: # pylint: disable=too-few-public-methods
         client = ModbusTcpClient()
         client.framer = Mock()
         client.register(CustomRequest)
-        assert client.framer.decoder.register.called_once_with(CustomRequest)
+        self.assertTrue(client.framer.decoder.register.called_once_with(CustomRequest))
 
     # -----------------------------------------------------------------------#
     # Test TLS Client
@@ -437,7 +437,7 @@ class CustomeRequest: # pylint: disable=too-few-public-methods
         client = ModbusTlsClient()
         client.framer = Mock()
         client.register(CustomeRequest)
-        assert client.framer.decoder.register.called_once_with(CustomeRequest)
+        self.assertTrue(client.framer.decoder.register.called_once_with(CustomeRequest))
 
     # -----------------------------------------------------------------------#
     # Test Serial Client
@@ -461,9 +461,9 @@ def test_sync_serial_client_instantiation(self):
     def test_sync_serial_rtu_client_timeouts(self): # pylint: disable=no-self-use
         """ Test sync serial rtu. """
         client = ModbusSerialClient(method="rtu", baudrate=9600)
-        assert client.silent_interval == round((3.5 * 11 / 9600), 6)
+        self.assertEqual(client.silent_interval, round((3.5 * 11 / 9600), 6))
         client = ModbusSerialClient(method="rtu", baudrate=38400)
-        assert client.silent_interval == round((1.75 / 1000), 6)
+        self.assertEqual(client.silent_interval, round((1.75 / 1000), 6))
 
     @patch("serial.Serial")
     def test_basic_sync_serial_client(self, mock_serial):

test/test_datastore.py

@@ -114,7 +114,7 @@ def test_modbus_sparse_data_block(self):
         block.setValues(0, {5: 32, 7: 43})
         self.assertEqual(block.getValues(5, 3), [32, 23, 43])
 
-        # assert value is empty dict when initialized without params
+        # assertEqual value is empty dict when initialized without params
         block = ModbusSparseDataBlock()
         self.assertEqual(block.values, {})
 
@@ -194,7 +194,7 @@ def test_str(self):
 
     def test_reset(self):
         """ Test reset. """
-        assert isinstance(self.slave.client, redis.Redis)
+        self.assertTrue(isinstance(self.slave.client, redis.Redis))
         self.slave.client = MagicMock()
         self.slave.reset()
         self.slave.client.flushall.assert_called_once_with()

test/test_factory.py

@@ -150,21 +150,21 @@ class CustomRequest(ModbusRequest):
             """ Custom request. """
             function_code = 0xff
         self.server.register(CustomRequest)
-        assert self.client.lookupPduClass(CustomRequest.function_code)
+        self.assertTrue(self.client.lookupPduClass(CustomRequest.function_code))
         CustomRequest.sub_function_code = 0xff
         self.server.register(CustomRequest)
-        assert self.server.lookupPduClass(CustomRequest.function_code)
+        self.assertTrue(self.server.lookupPduClass(CustomRequest.function_code))
 
     def test_client_register_custom_response(self):
         """ Test client register custom response. """
         class CustomResponse(ModbusResponse):
             """ Custom response. """
             function_code = 0xff
         self.client.register(CustomResponse)
-        assert self.client.lookupPduClass(CustomResponse.function_code)
+        self.assertTrue(self.client.lookupPduClass(CustomResponse.function_code))
         CustomResponse.sub_function_code = 0xff
         self.client.register(CustomResponse)
-        assert self.client.lookupPduClass(CustomResponse.function_code)
+        self.assertTrue(self.client.lookupPduClass(CustomResponse.function_code))
 #---------------------------------------------------------------------------#
 # I don't actually know what is supposed to be returned here, I assume that
 # since the high bit is set, it will simply echo the resulting message

test/test_server_asyncio.py

@@ -150,7 +150,7 @@ def eof_received(self):
                 host='127.0.0.1', port=random_port)
         await asyncio.wait_for(done, timeout=0.1)
 
-        assert received_value == expected_response
+        self.assertEqual(received_value, expected_response)
 
         transport.close()
         await asyncio.sleep(0)
@@ -182,13 +182,13 @@ def connection_made(self, transport):
         # On Windows we seem to need to give this an extra chance to finish,
         # otherwise there ends up being an active connection at the assert.
         await asyncio.sleep(0.2)
-        assert len(server.active_connections) == 1
+        self.assertEqual(len(server.active_connections), 1)
 
         protocol.transport.close()
             # close isn't synchronous and there's no
             # notification that it's done
         await asyncio.sleep(0.2)  # so we have to wait a bit
-        assert not server.active_connections
+        self.assertFalse(server.active_connections)
 
         server.server_close()
 
@@ -360,8 +360,8 @@ async def test_start_tls_server(self):
             identity = ModbusDeviceIdentification(info={0x00: 'VendorName'})
             self.loop = asynctest.Mock(self.loop) # pylint: disable=attribute-defined-outside-init
             server = await StartTlsServer(context=self.context, loop=self.loop, identity=identity)
-            assert server.control.Identity.VendorName == 'VendorName'
-            assert server.sslctx is not None
+            self.assertEqual(server.control.Identity.VendorName, 'VendorName')
+            self.assertTrue(server.sslctx is not None)
             self.loop.create_server.assert_called_once()
 
     async def test_tls_server_serve_forever(self):
@@ -414,11 +414,11 @@ async def test_udp_server_serve_forever_close(self):
 
         await server.serving
 
-        assert asyncio.isfuture(server.on_connection_terminated)
-        assert not server.on_connection_terminated.done()
+        self.assertTrue(asyncio.isfuture(server.on_connection_terminated))
+        self.assertFalse(server.on_connection_terminated.done())
 
         server.server_close()
-        assert server.protocol.is_closing()
+        self.assertTrue(server.protocol.is_closing())
 
     async def test_udp_server_serve_forever_twice(self):
         """ Call on serve_forever() twice should result in a runtime error """
@@ -521,7 +521,7 @@ def datagram_received(self, data, addr):
                                                 remote_addr=('127.0.0.1', random_port))
         await asyncio.wait_for(done, timeout=0.1)
 
-        assert received_value == expected_response
+        self.assertEqual(received_value, expected_response)
 
         transport.close()
         await asyncio.sleep(0)

test/test_server_context.py

@@ -56,8 +56,8 @@ def test_single_context_register(self): # pylint: disable=no-self-use
         request_db = [1, 2, 3]
         slave = ModbusSlaveContext()
         slave.register(0xff, 'custom_request', request_db)
-        assert slave.store["custom_request"] == request_db
-        assert slave.decode(0xff) == 'custom_request'
+        self.assertEqual(slave.store["custom_request"],request_db)
+        self.assertEqual(slave.decode(0xff),'custom_request')
 
 
 class ModbusServerMultipleContextTest(unittest.TestCase):