Robert Brewer avatar Robert Brewer committed 7f36758

More state tests (onStart/Stop tests). Also made it more explicit that onStop's are called even if there's no HTTP server.

Comments (0)

Files changed (3)

cherrypy/_cpserver.py

     def stop(self):
         """Stop, including any HTTP servers."""
         self.stop_http_server()
+        
+        for thread_ident, i in seen_threads.iteritems():
+            for func in self.onStopThreadList:
+                func(i)
+        seen_threads.clear()
+        
+        for func in self.onStopServerList:
+            func()
+        
         self.state = STOPPED
         cherrypy.log("CherryPy shut down", "HTTP")
     
             httpstop()
             cherrypy.log("HTTP Server shut down", "HTTP")
         
-        for thread_ident, i in seen_threads.iteritems():
-            for func in self.onStopThreadList:
-                func(i)
-        seen_threads.clear()
-        
-        for func in self.onStopServerList:
-            func()
-        
         self.httpserver = None
     
     def restart(self):

cherrypy/_cputil.py

     
     try:
         return globals()[name]
-    except KeyError:
+    except KeyError:
         msg = "Special attribute %s could not be found" % repr(name)
         raise cherrypy.HTTPError(500, msg)
 
         for i in xrange(len(objectList) - 1, -1, -1):
             if hasattr(objectList[i], name):
                 return "/" + "/".join(pathList[:i] + [name])
-    msg = "Special attribute %s could not be found" % repr(name)
+    msg = "Special attribute %s could not be found" % repr(name)
     raise cherrypy.HTTPError(500, msg)
 
 

cherrypy/test/test_states.py

     },
 })
 
+class Dependency:
+    
+    def __init__(self):
+        self.running = False
+        self.startcount = 0
+        self.threads = {}
+    
+    def start(self):
+        self.running = True
+        self.startcount += 1
+    
+    def stop(self):
+        self.running = False
+    
+    def startthread(self, thread_id):
+        self.threads[thread_id] = None
+    
+    def stopthread(self, thread_id):
+        del self.threads[thread_id]
+
 
 import helper
 
-
 class ServerStateTests(helper.CPWebCase):
     
     def test_0_NormalStateFlow(self):
         # get a NotReady error
         self.assertRaises(cherrypy.NotReady, self.getPage, "/")
         
+        # And our db_connection should not be running
+        self.assertEqual(db_connection.running, False)
+        self.assertEqual(db_connection.startcount, 0)
+        self.assertEqual(len(db_connection.threads), 0)
+        
         # Test server start
         cherrypy.server.start(True, self.serverClass)
         self.assertEqual(cherrypy.server.state, 1)
             port = cherrypy.config.get('server.socketPort')
             self.assertRaises(IOError, cherrypy._cpserver.check_port, host, port)
         
+        # The db_connection should be running now
+        self.assertEqual(db_connection.running, True)
+        self.assertEqual(db_connection.startcount, 1)
+        self.assertEqual(len(db_connection.threads), 0)
+        
         self.getPage("/")
         self.assertBody("Hello World")
+        self.assertEqual(len(db_connection.threads), 1)
         
         # Test server stop
         cherrypy.server.stop()
         
         # Once the server has stopped, we should get a NotReady error again.
         self.assertRaises(cherrypy.NotReady, self.getPage, "/")
+        
+        # Verify that the onStopServer function was called
+        self.assertEqual(db_connection.running, False)
+        self.assertEqual(len(db_connection.threads), 0)
     
     def test_1_Restart(self):
-        # Test server start
         cherrypy.server.start(True, self.serverClass)
+        
+        # The db_connection should be running now
+        self.assertEqual(db_connection.running, True)
+        sc = db_connection.startcount
+        
         self.getPage("/")
         self.assertBody("Hello World")
+        self.assertEqual(len(db_connection.threads), 1)
         
         # Test server restart from this thread
         cherrypy.server.restart()
         self.assertEqual(cherrypy.server.state, 1)
         self.getPage("/")
         self.assertBody("Hello World")
+        self.assertEqual(db_connection.running, True)
+        self.assertEqual(db_connection.startcount, sc + 1)
+        self.assertEqual(len(db_connection.threads), 1)
         
         # Test server restart from inside a page handler
         self.getPage("/restart")
         self.assertEqual(cherrypy.server.state, 1)
         self.assertBody("app was restarted succesfully")
+        self.assertEqual(db_connection.running, True)
+        self.assertEqual(db_connection.startcount, sc + 2)
+        # Since we are requesting synchronously, is only one thread used?
+        # Note that the "/restart" request has been flushed.
+        self.assertEqual(len(db_connection.threads), 0)
         
         cherrypy.server.stop()
         self.assertEqual(cherrypy.server.state, 0)
+        self.assertEqual(db_connection.running, False)
+        self.assertEqual(len(db_connection.threads), 0)
     
     def test_2_KeyboardInterrupt(self):
         if self.serverClass:
             self.assertEqual(cherrypy.server.httpserver, None)
             self.assertEqual(cherrypy.server.state, 0)
             self.assertRaises(cherrypy.NotReady, self.getPage, "/")
+            self.assertEqual(db_connection.running, False)
+            self.assertEqual(len(db_connection.threads), 0)
             
             # Raise a keyboard interrupt in a page handler; on multithreaded
             # servers, this should occur in one of the worker threads.
             self.assertEqual(cherrypy.server.httpserver, None)
             self.assertEqual(cherrypy.server.state, 0)
             self.assertRaises(cherrypy.NotReady, self.getPage, "/")
+            self.assertEqual(db_connection.running, False)
+            self.assertEqual(len(db_connection.threads), 0)
 
 
+db_connection = None
+
 def run(server, conf):
     helper.setConfig(conf)
     ServerStateTests.serverClass = server
     suite = helper.CPTestLoader.loadTestsFromTestCase(ServerStateTests)
     try:
+        global db_connection
+        db_connection = Dependency()
+        cherrypy.server.onStartServerList.append(db_connection.start)
+        cherrypy.server.onStopServerList.append(db_connection.stop)
+        cherrypy.server.onStartThreadList.append(db_connection.startthread)
+        cherrypy.server.onStopThreadList.append(db_connection.stopthread)
+        
         helper.CPTestRunner.run(suite)
     finally:
         cherrypy.server.stop()
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.