Commits

Robert Brewer committed 292d7ef

New helper.CPProcess class (moved from test_states), plus a bugfix to allow Ctrl-C to stop CP when running tests in --server mode.

Comments (0)

Files changed (2)

cherrypy/test/helper.py

 # requirements. So don't go moving functions from here into test.py,
 # or vice-versa, unless you *really* know what you're doing.
 
+import os
+thisdir = os.path.abspath(os.path.dirname(__file__))
 import re
 import sys
 import thread
+import time
 import warnings
 
 import cherrypy
 
 def testmain(conf=None):
     """Run __main__ as a test module, with webtest debugging."""
+    engine = cherrypy.engine
     if '--server' in sys.argv:
         # Run the test module server-side only; wait for Ctrl-C to break.
         conf = conf or {}
         conf['server.socket_host'] = '0.0.0.0'
         setConfig(conf)
-        cherrypy.engine.start()
-        cherrypy.engine.block()
+        if hasattr(engine, "signal_handler"):
+            engine.signal_handler.subscribe()
+        if hasattr(engine, "console_control_handler"):
+            engine.console_control_handler.subscribe()
+        engine.start()
+        engine.block()
     else:
         for arg in sys.argv:
             if arg.startswith('--client='):
             conf = conf or {}
             conf['server.socket_host'] = '127.0.0.1'
             setConfig(conf)
-            cherrypy.engine.start_with_callback(_test_main_thread)
-            cherrypy.engine.block()
+            engine.start_with_callback(_test_main_thread)
+            engine.block()
 
 def _test_main_thread():
     try:
     finally:
         cherrypy.engine.exit()
 
+
+
+# --------------------------- Spawning helpers --------------------------- #
+
+
+class CPProcess(object):
+    
+    pid_file = os.path.join(thisdir, 'test.pid')
+    config_file = os.path.join(thisdir, 'test.conf')
+    config_template = """[global]
+server.socket_host: '%(host)s'
+server.socket_port: %(port)s
+log.screen: False
+log.error_file: r'%(error_log)s'
+log.access_file: r'%(access_log)s'
+%(ssl)s
+%(extra)s
+"""
+    error_log = os.path.join(thisdir, 'test.error.log')
+    access_log = os.path.join(thisdir, 'test.access.log')
+    
+    def __init__(self, wait=False, daemonize=False, ssl=False):
+        self.wait = wait
+        self.daemonize = daemonize
+        self.ssl = ssl
+        self.host = cherrypy.server.socket_host
+        self.port = cherrypy.server.socket_port
+    
+    def write_conf(self, extra=""):
+        if self.ssl:
+            serverpem = os.path.join(thisdir, 'test.pem')
+            ssl = """
+server.ssl_certificate: r'%s'
+server.ssl_private_key: r'%s'
+""" % (serverpem, serverpem)
+        else:
+            ssl = ""
+        
+        f = open(self.config_file, 'wb')
+        f.write(self.config_template %
+                {'host': self.host,
+                 'port': self.port,
+                 'error_log': self.error_log,
+                 'access_log': self.access_log,
+                 'ssl': ssl,
+                 'extra': extra,
+                 })
+        f.close()
+    
+    def start(self, imports=None):
+        """Start cherryd in a subprocess."""
+        cherrypy._cpserver.wait_for_free_port(self.host, self.port)
+        
+        args = [sys.executable, os.path.join(thisdir, '..', 'cherryd'),
+                '-c', self.config_file, '-p', self.pid_file]
+        
+        if not isinstance(imports, (list, tuple)):
+            imports = [imports]
+        for i in imports:
+            if i:
+                args.append('-i')
+                args.append(i)
+        
+        if self.daemonize:
+            args.append('-d')
+        
+        if self.wait:
+            self.exit_code = os.spawnl(os.P_WAIT, sys.executable, *args)
+        else:
+            os.spawnl(os.P_NOWAIT, sys.executable, *args)
+            cherrypy._cpserver.wait_for_occupied_port(self.host, self.port)
+        
+        # Give the engine a wee bit more time to finish STARTING
+        if self.daemonize:
+            time.sleep(2)
+        else:
+            time.sleep(1)
+    
+    def get_pid(self):
+        return int(open(self.pid_file, 'rb').read())
+    
+    def join(self):
+        """Wait for the process to exit."""
+        try:
+            try:
+                # Mac, UNIX
+                os.wait()
+            except AttributeError:
+                # Windows
+                try:
+                    pid = self.get_pid()
+                except IOError:
+                    # Assume the subprocess deleted the pidfile on shutdown.
+                    pass
+                else:
+                    os.waitpid(pid, 0)
+        except OSError, x:
+            if x.args != (10, 'No child processes'):
+                raise
+

cherrypy/test/test_states.py

 import cherrypy
 engine = cherrypy.engine
 thisdir = os.path.join(os.getcwd(), os.path.dirname(__file__))
-PID_file_path = os.path.join(thisdir, 'pid_for_test_daemonize')
-
-
-def write_conf(scheme='http', extra=""):
-    if scheme.lower() == 'https':
-        serverpem = os.path.join(thisdir, 'test.pem')
-        ssl = """
-server.ssl_certificate: r'%s'
-server.ssl_private_key: r'%s'
-""" % (serverpem, serverpem)
-    else:
-        ssl = ""
-    
-    conffile = open(os.path.join(thisdir, 'test_states.conf'), 'wb')
-    conffile.write("""[global]
-server.socket_host: '%(host)s'
-server.socket_port: %(port)s
-log.screen: False
-log.error_file: r'%(error_log)s'
-log.access_file: r'%(access_log)s'
-%(ssl)s
-%(extra)s
-""" % {'host': host,
-       'port': port,
-       'error_log': os.path.join(thisdir, 'test_states_demo.error.log'),
-       'access_log': os.path.join(thisdir, 'test_states_demo.access.log'),
-       'ssl': ssl,
-       'extra': extra,
-       })
-    conffile.close()
-
-
-def spawn_cp(configfile=os.path.join(thisdir, 'test_states.conf'),
-             wait=False, daemonize=False):
-    """Start cherryd in a subprocess."""
-    host = cherrypy.server.socket_host
-    port = cherrypy.server.socket_port
-    cherrypy._cpserver.wait_for_free_port(host, port)
-    
-    args = [sys.executable, os.path.join(thisdir, '..', 'cherryd'),
-            '-c', configfile, '-i', 'cherrypy.test.test_states_demo']
-    
-    if sys.platform != 'win32':
-        args.append('-p')
-        args.append(PID_file_path)
-    
-    # Spawn the process and wait, when this returns, the original process
-    # is finished.  If it daemonized properly, we should still be able
-    # to access pages.
-    if daemonize:
-        args.append('-d')
-    
-    if wait:
-        result = os.spawnl(os.P_WAIT, sys.executable, *args)
-    else:
-        result = os.spawnl(os.P_NOWAIT, sys.executable, *args)
-        cherrypy._cpserver.wait_for_occupied_port(host, port)
-    
-    # Give the engine a wee bit more time to finish STARTING
-    if daemonize:
-        time.sleep(2)
-    else:
-        time.sleep(1)
-    
-    return result
-
-def wait(pid):
-    """Wait for the process with the given pid to exit."""
-    try:
-        try:
-            # Mac, UNIX
-            os.wait()
-        except AttributeError:
-            # Windows
-            os.waitpid(pid, 0)
-    except OSError, x:
-        if x.args != (10, 'No child processes'):
-            raise
 
 
 class Root:
             return
         
         # Start the demo script in a new process
-        write_conf(scheme=self.scheme)
-        pid = spawn_cp()
+        p = helper.CPProcess(ssl=(self.scheme.lower()=='https'))
+        p.write_conf()
+        p.start(imports='cherrypy.test.test_states_demo')
         try:
             self.getPage("/start")
             start = float(self.body)
             time.sleep(2)
             cherrypy._cpserver.wait_for_occupied_port(host, port)
             
-            self.getPage("/pid")
-            pid = int(self.body)
-            
             self.getPage("/start")
             self.assert_(float(self.body) > start)
         finally:
             # Shut down the spawned process
             self.getPage("/exit")
-        wait(pid)
+        p.join()
     
     def test_5_Start_Error(self):
         if not self.server_class:
         
         # If a process errors during start, it should stop the engine
         # and exit with a non-zero exit code.
-        write_conf(scheme=self.scheme, extra="starterror: True")
-        exit_code = spawn_cp(wait=True)
-        if exit_code == 0:
+        p = helper.CPProcess(ssl=(self.scheme.lower()=='https'),
+                             wait=True)
+        p.write_conf(extra="starterror: True")
+        p.start(imports='cherrypy.test.test_states_demo')
+        if p.exit_code == 0:
             self.fail("Process failed to return nonzero exit code.")
 
 
         # Spawn the process and wait, when this returns, the original process
         # is finished.  If it daemonized properly, we should still be able
         # to access pages.
-        write_conf(scheme=self.scheme)
-        exit_code = spawn_cp(wait=True, daemonize=True)
-        
-        # Get the PID from the file.
-        pid = int(open(PID_file_path).read())
+        p = helper.CPProcess(ssl=(self.scheme.lower()=='https'),
+                             wait=True, daemonize=True)
+        p.write_conf()
+        p.start(imports='cherrypy.test.test_states_demo')
         try:
             # Just get the pid of the daemonization process.
             self.getPage("/pid")
             self.assertStatus(200)
             page_pid = int(self.body)
-            self.assertEqual(page_pid, pid)
+            self.assertEqual(page_pid, p.get_pid())
         finally:
             # Shut down the spawned process
             self.getPage("/exit")
-        wait(pid)
+        p.join()
         
         # Wait until here to test the exit code because we want to ensure
         # that we wait for the daemon to finish running before we fail.
-        if exit_code != 0:
+        if p.exit_code != 0:
             self.fail("Daemonized parent process failed to exit cleanly.")
 
 
             return
         
         # Spawn the process.
-        write_conf(scheme=self.scheme)
-        pid = spawn_cp(wait=False, daemonize=False)
+        p = helper.CPProcess(ssl=(self.scheme.lower()=='https'))
+        p.write_conf()
+        p.start(imports='cherrypy.test.test_states_demo')
         # Send a SIGHUP
-        os.kill(pid, SIGHUP)
+        os.kill(p.get_pid(), SIGHUP)
         # This might hang if things aren't working right, but meh.
-        wait(pid)
+        p.join()
     
     def test_SIGHUP_daemonized(self):
         # When daemonized, SIGHUP should restart the server.
         # Spawn the process and wait, when this returns, the original process
         # is finished.  If it daemonized properly, we should still be able
         # to access pages.
-        write_conf(scheme=self.scheme)
-        exit_code = spawn_cp(wait=True, daemonize=True)
+        p = helper.CPProcess(ssl=(self.scheme.lower()=='https'),
+                             wait=True, daemonize=True)
+        p.write_conf()
+        p.start(imports='cherrypy.test.test_states_demo')
         
-        # Get the PID from the file.
-        pid = int(open(PID_file_path).read())
+        pid = p.get_pid()
         try:
             # Send a SIGHUP
             os.kill(pid, SIGHUP)
         finally:
             # Shut down the spawned process
             self.getPage("/exit")
-        wait(new_pid)
+        p.join()
     
     def test_SIGTERM(self):
         # SIGTERM should shut down the server whether daemonized or not.
             return
         
         # Spawn a normal, undaemonized process.
-        write_conf(scheme=self.scheme)
-        pid = spawn_cp(wait=False, daemonize=False)
+        p = helper.CPProcess(ssl=(self.scheme.lower()=='https'))
+        p.write_conf()
+        p.start(imports='cherrypy.test.test_states_demo')
         # Send a SIGTERM
-        os.kill(pid, SIGTERM)
+        os.kill(p.get_pid(), SIGTERM)
         # This might hang if things aren't working right, but meh.
-        wait(pid)
+        p.join()
         
         if os.name in ['posix']: 
             # Spawn a daemonized process and test again.
-            exit_code = spawn_cp(wait=True, daemonize=True)
+            p = helper.CPProcess(ssl=(self.scheme.lower()=='https'),
+                                 wait=True, daemonize=True)
+            p.write_conf()
+            p.start(imports='cherrypy.test.test_states_demo')
             # Send a SIGTERM
-            pid = int(open(PID_file_path).read())
-            os.kill(pid, SIGTERM)
+            os.kill(p.get_pid(), SIGTERM)
             # This might hang if things aren't working right, but meh.
-            wait(pid)
+            p.join()
     
     def test_signal_handler_unsubscribe(self):
         if not self.server_class:
             return
         
         # Spawn a normal, undaemonized process.
-        write_conf(scheme=self.scheme, extra="unsubsig: True")
-        pid = spawn_cp(wait=False, daemonize=False)
-        os.kill(pid, SIGTERM)
-        wait(pid)
+        p = helper.CPProcess(ssl=(self.scheme.lower()=='https'))
+        p.write_conf(extra="unsubsig: True")
+        p.start(imports='cherrypy.test.test_states_demo')
+        # Send a SIGTERM
+        os.kill(p.get_pid(), SIGTERM)
+        # This might hang if things aren't working right, but meh.
+        p.join()
+        
         # Assert the old handler ran.
-        errlog = os.path.join(thisdir, 'test_states_demo.error.log')
-        target_line = open(errlog, 'rb').readlines()[-10]
+        target_line = open(p.error_log, 'rb').readlines()[-10]
         if not "I am an old SIGTERM handler." in target_line:
             self.fail("Old SIGTERM handler did not run.\n%r" % target_line)
 
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.