Use pytest_configure instead of import directly. (#754)

Author: janiversen

test/conftest.py

@@ -1,15 +1,17 @@
+import pytest
 import platform
 from pkg_resources import parse_version
 
 
-IS_DARWIN = platform.system().lower() == "darwin"
-IS_WINDOWS = platform.system().lower() == "windows"
+def pytest_configure():
+    pytest.IS_DARWIN = platform.system().lower() == "darwin"
+    pytest.IS_WINDOWS = platform.system().lower() == "windows"
 
-if IS_DARWIN:
-    # check for SIERRA
-    if parse_version("10.12") < parse_version(platform.mac_ver()[0]):
-        SERIAL_PORT = '/dev/ptyp0'
+    if pytest.IS_DARWIN:
+        # check for SIERRA
+        if parse_version("10.12") < parse_version(platform.mac_ver()[0]):
+            pytest.SERIAL_PORT = '/dev/ptyp0'
+        else:
+            pytest.SERIAL_PORT = '/dev/ttyp0'
     else:
-        SERIAL_PORT = '/dev/ttyp0'
-else:
-    SERIAL_PORT = "/dev/ptmx"
+        pytest.SERIAL_PORT = "/dev/ptmx"

test/test_client_async.py

@@ -25,7 +25,6 @@
 from pymodbus.transaction import ModbusSocketFramer, ModbusTlsFramer, ModbusRtuFramer
 from pymodbus.transaction import ModbusAsciiFramer, ModbusBinaryFramer
 from pymodbus.client.asynchronous.twisted import ModbusSerClientProtocol
-from test.conftest import SERIAL_PORT
 
 import ssl
 
@@ -191,7 +190,7 @@ def testSerialTwistedClient(self, method, framer):
 
                     protocol, client = AsyncModbusSerialClient(schedulers.REACTOR,
                                                                method=method,
-                                                               port=SERIAL_PORT,
+                                                               port=pytest.SERIAL_PORT,
                                                                proto_cls=ModbusSerClientProtocol)
 
                     assert (isinstance(client, SerialPort))
@@ -219,12 +218,12 @@ def testSerialTornadoClient(self, method, framer):
         """ Test the serial tornado client client initialize """
         from serial import Serial
         with maybe_manage(sys.platform in ('darwin', 'win32'), patch.object(Serial, "open")):
-            protocol, future = AsyncModbusSerialClient(schedulers.IO_LOOP, method=method, port=SERIAL_PORT)
+            protocol, future = AsyncModbusSerialClient(schedulers.IO_LOOP, method=method, port=pytest.SERIAL_PORT)
             client = future.result()
             assert(isinstance(client, AsyncTornadoModbusSerialClient))
             assert(0 == len(list(client.transaction)))
             assert(isinstance(client.framer, framer))
-            assert(client.port == SERIAL_PORT)
+            assert(client.port == pytest.SERIAL_PORT)
             assert(client._connected)
 
             def handle_failure(failure):
@@ -251,11 +250,11 @@ def testSerialAsyncioClient(self, mock_gather, mock_event_loop, method, framer):
         """
         loop = asyncio.get_event_loop()
         loop.is_running.side_effect = lambda: False
-        loop, client = AsyncModbusSerialClient(schedulers.ASYNC_IO, method=method, port=SERIAL_PORT, loop=loop,
+        loop, client = AsyncModbusSerialClient(schedulers.ASYNC_IO, method=method, port=pytest.SERIAL_PORT, loop=loop,
                                                baudrate=19200, parity='E', stopbits=2, bytesize=7)
         assert(isinstance(client, AsyncioModbusSerialClient))
         assert(isinstance(client.framer, framer))
-        assert(client.port == SERIAL_PORT)
+        assert(client.port == pytest.SERIAL_PORT)
         assert(client.baudrate == 19200)
         assert(client.parity == 'E')
         assert(client.stopbits == 2)

test/test_client_async_tornado.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python3
 import unittest
 from unittest.mock import patch, Mock
+import pytest
 
 from pymodbus.client.asynchronous.tornado import (BaseTornadoClient,
                                                   AsyncModbusSerialClient, AsyncModbusUDPClient, AsyncModbusTCPClient
@@ -11,7 +12,6 @@
 from pymodbus.exceptions import ConnectionException
 from pymodbus.transaction import ModbusSocketFramer, ModbusRtuFramer
 from pymodbus.bit_read_message import ReadCoilsRequest, ReadCoilsResponse
-from test.conftest import SERIAL_PORT
 
 # ---------------------------------------------------------------------------#
 # Fixture
@@ -166,7 +166,7 @@ def testSerialClientInit(self):
         client = AsyncModbusSerialClient(ioloop=schedulers.IO_LOOP,
                                          framer=ModbusRtuFramer(
                                              ClientDecoder()),
-                                         port=SERIAL_PORT)
+                                         port=pytest.SERIAL_PORT)
         self.assertEqual(0, len(list(client.transaction)))
         self.assertTrue(isinstance(client.framer, ModbusRtuFramer))
 
@@ -182,8 +182,8 @@ def testSerialClientConnect(self, mock_serial, mock_seriostream, mock_ioloop):
         client = AsyncModbusSerialClient(ioloop=schedulers.IO_LOOP,
                                          framer=ModbusRtuFramer(
                                              ClientDecoder()),
-                                         port=SERIAL_PORT)
-        self.assertTrue(client.port, SERIAL_PORT)
+                                         port=pytest.SERIAL_PORT)
+        self.assertTrue(client.port, pytest.SERIAL_PORT)
         self.assertFalse(client._connected)
         client.connect()
         self.assertTrue(client._connected)
@@ -197,7 +197,7 @@ def testSerialClientDisconnect(self, mock_serial, mock_seriostream, mock_ioloop)
         client = AsyncModbusSerialClient(ioloop=schedulers.IO_LOOP,
                                          framer=ModbusRtuFramer(
                                              ClientDecoder()),
-                                         port=SERIAL_PORT)
+                                         port=pytest.SERIAL_PORT)
         client.connect()
         self.assertTrue(client._connected)
 
@@ -217,7 +217,7 @@ def testSerialClientExecute(self, mock_serial, mock_seriostream, mock_ioloop):
         client = AsyncModbusSerialClient(ioloop=schedulers.IO_LOOP,
                                          framer=ModbusRtuFramer(
                                              ClientDecoder()),
-                                         port=SERIAL_PORT,
+                                         port=pytest.SERIAL_PORT,
                                          timeout=0)
         client.connect()
         client.stream = Mock()
@@ -237,7 +237,7 @@ def testSerialClientHandleResponse(self, mock_serial, mock_seriostream, mock_iol
         client = AsyncModbusSerialClient(ioloop=schedulers.IO_LOOP,
                                          framer=ModbusRtuFramer(
                                              ClientDecoder()),
-                                         port=SERIAL_PORT)
+                                         port=pytest.SERIAL_PORT)
         client.connect()
         out = []
         reply = ReadCoilsRequest(1, 1)
@@ -261,7 +261,7 @@ def testSerialClientBuildResponse(self, mock_serial, mock_seriostream, mock_iolo
         client = AsyncModbusSerialClient(ioloop=schedulers.IO_LOOP,
                                          framer=ModbusRtuFramer(
                                              ClientDecoder()),
-                                         port=SERIAL_PORT)
+                                         port=pytest.SERIAL_PORT)
         self.assertEqual(0, len(list(client.transaction)))
 
         def handle_failure(failure):

test/test_server_async.py

@@ -14,7 +14,6 @@
 from pymodbus.compat import byte2int
 from pymodbus.transaction import ModbusSocketFramer
 from pymodbus.exceptions import NoSuchSlaveException, ModbusIOException
-from test.conftest import SERIAL_PORT
 
 import sys
 # --------------------------------------------------------------------------- #
@@ -183,7 +182,7 @@ def testUdpServerStartup(self):
     def testSerialServerStartup(self, mock_sp):
         ''' Test that the modbus serial asynchronous server starts correctly '''
         with patch('twisted.internet.reactor') as mock_reactor:
-            StartSerialServer(context=None, port=SERIAL_PORT)
+            StartSerialServer(context=None, port=pytest.SERIAL_PORT)
             self.assertEqual(mock_reactor.run.call_count, 1)
 
     @no_twisted_serial_on_windows_with_pypy
@@ -194,7 +193,7 @@ def testStopServerFromMainThread(self, mock_sp):
         :return:
         """
         with patch('twisted.internet.reactor') as mock_reactor:
-            StartSerialServer(context=None, port=SERIAL_PORT)
+            StartSerialServer(context=None, port=pytest.SERIAL_PORT)
             self.assertEqual(mock_reactor.run.call_count, 1)
             StopServer()
             self.assertEqual(mock_reactor.stop.call_count, 1)
@@ -209,7 +208,7 @@ def testStopServerFromThread(self, mock_sp):
         from threading import Thread
         import time
         with patch('twisted.internet.reactor') as mock_reactor:
-            StartSerialServer(context=None, port=SERIAL_PORT)
+            StartSerialServer(context=None, port=pytest.SERIAL_PORT)
             self.assertEqual(mock_reactor.run.call_count, 1)
             t = Thread(target=StopServer)
             t.start()

test/test_server_asyncio.py

@@ -17,7 +17,6 @@
 
 
 from pymodbus.exceptions import NoSuchSlaveException
-from test.conftest import IS_WINDOWS, SERIAL_PORT
 
 # ---------------------------------------------------------------------------#
 # Fixture
@@ -112,7 +111,7 @@ def eof_received(self):
             self.assertTrue( process.call_args[1]["data"] == data )
             server.server_close()
 
-    @pytest.mark.skipif(IS_WINDOWS, reason="To fix")    
+    @pytest.mark.skipif(pytest.IS_WINDOWS, reason="To fix")    
     async def testTcpServerRoundtrip(self):
         ''' Test sending and receiving data on tcp socket '''
         data = b"\x01\x00\x00\x00\x00\x06\x01\x03\x00\x00\x00\x01" # unit 1, read register
@@ -150,7 +149,7 @@ def eof_received(self):
         await asyncio.sleep(0)
         server.server_close()
 
-    @pytest.mark.skipif(IS_WINDOWS, reason="To fix")    
+    @pytest.mark.skipif(pytest.IS_WINDOWS, reason="To fix")    
     async def testTcpServerConnectionLost(self):
         """ Test tcp stream interruption """
 

test/test_server_sync.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python3
 import unittest
 from unittest.mock import patch, Mock
-
+import pytest
 import serial
 import socket
 import ssl
@@ -20,7 +20,6 @@
 from pymodbus.transaction import ModbusTlsFramer
 
 from pymodbus.compat import socketserver
-from test.conftest import SERIAL_PORT
 
 
 # --------------------------------------------------------------------------- #
@@ -400,7 +399,7 @@ def testStartUdpServer(self):
     def testStartSerialServer(self):
         ''' Test the serial server starting factory '''
         with patch.object(ModbusSerialServer, 'serve_forever') as mock_server:
-            StartSerialServer(port=SERIAL_PORT)
+            StartSerialServer(port=pytest.SERIAL_PORT)
 
 
 # --------------------------------------------------------------------------- #