Clean up pytest fixtures for better Pypy3 compatibility

o Explicitly kill server fixtures; still not quite working?

Author: pjkundert

modbus_test.py

@@ -14,13 +14,16 @@
 import subprocess
 import sys
 import time
+import traceback
+
+log				= logging.getLogger( 'mbs_test' )
 
 has_o_nonblock			= False
 try:
     import fcntl
     has_o_nonblock		= True
 except Exception:
-    logging.warning( "Failed to import fcntl; skipping simulated Modbus/TCP PLC tests" )
+    log.warning( "Failed to import fcntl; skipping simulated Modbus/TCP PLC tests" )
 
 from . import misc
 from .tools.waits import waitfor
@@ -39,15 +42,15 @@ class nonblocking_command( object ):
 
             try:
                 data 		= command.stdout.read()
-                logging.debug( "Received %d bytes from command, len( data ))
+                log.debug( "Received %d bytes from command, len( data ))
                 collect        += data
             except IOError as exc:
                 if exc.errno != errno.EAGAIN:
-                    logging.warning( "I/O Error reading data: %s" % traceback.format_exc() )
+                    log.warning( "I/O Error reading data: %s" % traceback.format_exc() )
                     command	= None
                 # Data not presently available; ignore
             except:
-                logging.warning( "Exception reading data: %s", traceback.format_exc() )
+                log.warning( "Exception reading data: %s", traceback.format_exc() )
                 command		= None
 
             # do other stuff in loop...
@@ -62,32 +65,34 @@ class nonblocking_command( object ):
     def __init__( self, command, stderr=subprocess.STDOUT, stdin=None, bufsize=0, blocking=None ):
         shell			= type( command ) is not list
         self.command		= ' '.join( command ) if not shell else command
-        logging.info( "Starting command: %s", self.command )
+        log.info( "Starting command: %s", self.command )
         self.process		= subprocess.Popen(
             command, stdout=subprocess.PIPE, stderr=stderr, stdin=stdin,
             bufsize=bufsize, preexec_fn=os.setsid, shell=shell )
+        log.normal( 'Started Server PID [%d]: %s', self.process.pid, self.command )
         if not blocking:
             fd 			= self.process.stdout.fileno()
             fl			= fcntl.fcntl( fd, fcntl.F_GETFL )
             fcntl.fcntl( fd, fcntl.F_SETFL, fl | os.O_NONBLOCK )
-
-        atexit.register( lambda: self.kill() )
+        # Really, really ensure we get terminated
+        atexit.register( self.kill )
 
     @property
     def stdout( self ):
         return self.process.stdout
 
     def kill( self ):
-        logging.info( 'Sending SIGTERM to PID [%d]: %s', self.process.pid, self.command )
+        log.normal( 'Sending SIGTERM to PID [%d]: %s, via: %s', self.process.pid, self.command,
+                        ''.join( traceback.format_stack() ) if log.isEnabledFor( logging.INFO ) else '' )
         try:
             os.killpg( self.process.pid, signal.SIGTERM )
         except OSError as exc:
-            logging.info( 'Failed to send SIGTERM to PID [%d]: %s', self.process.pid, exc )
+            log.info( 'Failed to send SIGTERM to PID [%d]: %s', self.process.pid, exc )
         else:
-            logging.info( "Waiting for command (PID [%d]) to terminate", self.process.pid )
+            log.info( "Waiting for command (PID [%d]) to terminate", self.process.pid )
             self.process.wait()
         # Process may exit with a non-numeric returncode (eg. None)
-        logging.info( "Command (PID [%d]) finished with status %r: %s",
+        log.info( "Command (PID [%d]) finished with status %r: %s",
                       self.process.pid, self.process.returncode, self.command )
 
     __del__			= kill
@@ -111,30 +116,30 @@ def start_modbus_simulator( options ):
         raw			= None
         try:
             raw			= command.stdout.read()
-            logging.debug( "Socket received: %r", raw)
+            log.debug( "Socket received: %r", raw)
             if raw:
                 data  	       += raw.decode( 'utf-8', 'backslashreplace' )
         except IOError as exc:
-            logging.debug( "Socket blocking...")
+            log.debug( "Socket blocking...")
             assert exc.errno == errno.EAGAIN, "Expected only Non-blocking IOError"
         except Exception as exc:
-            logging.warning("Socket read return Exception: %s", exc)
+            log.warning("Socket read return Exception: %s", exc)
         if not raw:
             time.sleep( RTU_LATENCY )
         while data.find( '\n' ) >= 0:
             line,data		= data.split( '\n', 1 )
-            logging.info( "%s", line )
+            log.info( "%s", line )
             m			= re.search( "address = (.*)", line )
             if m:
                 try:
                     host,port	= m.group(1).split( ':' )
                     address	= host,int(port)
-                    logging.normal( "Modbus/TCP Simulator started after %7.3fs on %s:%d",
+                    log.normal( "Modbus/TCP Simulator started after %7.3fs on %s:%d",
                                     misc.timer() - begun, address[0], address[1] )
                 except:
                     assert m.group(1).startswith( '/' )
                     address	= m.group(1)
-                    logging.normal( "Modbus/RTU Simulator started after %7.3fs on %s",
+                    log.normal( "Modbus/RTU Simulator started after %7.3fs on %s",
                                     misc.timer() - begun, address )
                 break
     return command,address
@@ -209,7 +214,7 @@ def run_plc_modbus_polls( plc ):
             r			= random.randint( base, base + extent )
         v			= plc.read( r )
         if v is not None:
-            logging.detail( "New reg %5d was already polled due to reach=%d", r, plc.reach )
+            log.detail( "New reg %5d was already polled due to reach=%d", r, plc.reach )
             regs[r]		= v
         regs[r]			= ( regs[r] ^ 1 if r in regs
                                 else random.randint( 0, 65535 ) if base > 40000
@@ -225,7 +230,7 @@ def run_plc_modbus_polls( plc ):
             assert success
             rolling		= misc.exponential_moving_average( rolling, elapsed, rolling_factor )
 
-        logging.normal( "%3d/%3d regs: polled %3d ranges w/in %7.3fs. Polled %5d == %5d w/in %7.3fs: avg. %7.3fs (load %3.2f, %3.2f, %3.2f)",
+        log.normal( "%3d/%3d regs: polled %3d ranges w/in %7.3fs. Polled %5d == %5d w/in %7.3fs: avg. %7.3fs (load %3.2f, %3.2f, %3.2f)",
                          len( regs ), total, len( plc.polling ), plc.duration,
                          r, regs[r], elapsed or 0.0, rolling or 0.0, *[misc.nan if load is None else load for load in plc.load] )
 

remote_test.py

@@ -52,17 +52,20 @@
 except ImportError:
     logging.warning( "Failed to import pymodbus module; skipping Modbus/TCP related tests; run 'pip install pymodbus'" )
 
+
 @pytest.fixture( scope="module" )
-def simulated_modbus_tcp():
+def simulated_modbus_tcp( request ):
     """Start a simulator over a range of ports; parse the port successfully bound."""
-    return start_modbus_simulator( options=[
+    command,address		= start_modbus_simulator( options=[
         '-vv', '--log', 'remote_test.modbus_sim.log.localhost:11502',
         '--evil', 'delay:.25',
         '--address', 'localhost:11502',
         '--range', '10',
         '    1 -  1000 = 0',
         '40001 - 41000 = 0',
     ] )
+    request.addfinalizer( command.kill )
+    return command,address
 
 
 @pytest.mark.skipif( not has_pymodbus, reason="Needs pymodbus" )

serial_test.py

@@ -141,13 +141,20 @@ def simulated_modbus_rtu( tty ):
         } )
     ] )
 
+
 @pytest.fixture( scope="module" )
-def simulated_modbus_rtu_ttyS0():
-    return simulated_modbus_rtu( "/dev/ttyS0" )
+def simulated_modbus_rtu_ttyS0( request ):
+    command,address		= simulated_modbus_rtu( "/dev/ttyS0" )
+    request.addfinalizer( command.kill )
+    return command,address
+
 
 @pytest.fixture( scope="module" )
 def simulated_modbus_rtu_ttyS2():
-    return simulated_modbus_rtu( "/dev/ttyS2" )
+    command,address		= simulated_modbus_rtu( "/dev/ttyS2" )
+    request.addfinalizer( command.kill )
+    return command,address
+
 
 @pytest.mark.skipif( 'SERIAL_TEST' not in os.environ or not has_o_nonblock or not has_minimalmodbus or not has_pyserial,
                      reason="Needs SERIAL_TEST and fcntl/O_NONBLOCK and minimalmodbus and pyserial" )

server/enip/hart_test.py

@@ -100,8 +100,11 @@ def command_logging( command, buf='' ):
 
 
 @pytest.fixture( scope="module" )
-def simulated_hart_gateway():
-    return start_hart_simulator()
+def simulated_hart_gateway( request ):
+    command,address		= start_hart_simulator()
+    request.addfinalizer( command.kill )
+    return command,address
+
 
 hart_kwds			= dict(
     timeout		= 15.0,

server/enip/poll_test.py

@@ -14,6 +14,7 @@
 import sys
 import time
 import threading
+import traceback
 
 import pytest
 
@@ -94,8 +95,10 @@ def start_powerflex_simulator( *options, **kwds ):
 
 
 @pytest.fixture( scope="module" )
-def simulated_powerflex_gateway():
-    return start_powerflex_simulator( 'SCADA=INT[100]' )
+def simulated_powerflex_gateway( request ):
+    command,address		= start_powerflex_simulator( 'SCADA=INT[100]' )
+    request.addfinalizer( command.kill )
+    return command,address
 
 
 def test_powerflex_simple( simulated_powerflex_gateway ):

server/enip/udt_test.py

@@ -134,7 +134,7 @@ def test_logix_remote_udt( count=1 ):
     """
     #logging.getLogger().setLevel( logging.NORMAL )
     device.lookup_reset() # Flush out any existing CIP Objects for a fresh start
-    svraddr		        = ('localhost', 44818)
+    svraddr		        = ('localhost', 44838)
 
     tagname			= example_tagname
     tagaddr			= example_enip_addr
@@ -287,6 +287,8 @@ def logix_remote_udt_pylogix( count, svraddr, kwargs ):
             comm.SocketTimeout	= timeout
             comm.IPAddress	= svraddr[0]
 
+            comm.conn.Port	= int( svraddr[1] )
+
             # CIP Register, Forward Open
             start		= misc.timer()
             conn		= comm.conn.connect()

server/logix_test.py

@@ -806,7 +806,7 @@ def test_logix_remote_pylogix( count=100 ):
     """
     #logging.getLogger().setLevel( logging.NORMAL )
     enip.lookup_reset() # Flush out any existing CIP Objects for a fresh start
-    svraddr		        = ('localhost', 44818)
+    svraddr		        = ('localhost', 44828)
     kwargs			= {
         'argv': [
             #'-v',
@@ -856,6 +856,8 @@ def logix_remote_pylogix( count, svraddr, kwargs ):
         comm.IPAddress		= svraddr[0]
         comm.ConnectionSize	= 4000
 
+        comm.conn.Port		= int( svraddr[1] )
+
         # CIP Register, Forward Open
         start			= cpppo.timer()
         conn			= comm.conn.connect()