import httplib from httplib import BadStatusLine import os import sys import threading import time from cherrypy.test import test test.prefer_parent_path() import cherrypy class Root: def index(self): return "Hello World" index.exposed = True def ctrlc(self): raise KeyboardInterrupt() ctrlc.exposed = True def restart(self): cherrypy.engine.restart() return "app was restarted succesfully" restart.exposed = True def block_explicit(self): while True: if cherrypy.response.timed_out: cherrypy.response.timed_out = False return "broken!" time.sleep(0.01) block_explicit.exposed = True def block_implicit(self): time.sleep(0.5) return "response.timeout = %s" % cherrypy.response.timeout block_implicit.exposed = True cherrypy.tree.mount(Root()) cherrypy.config.update({ 'environment': 'test_suite', 'engine.deadlock_poll_freq': 0.1, 'response.timeout': 0.2, }) 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] from cherrypy.test import helper class ServerStateTests(helper.CPWebCase): def test_0_NormalStateFlow(self): if not self.server_class: # Without having called "cherrypy.engine.start()", we should # get a 503 Service Unavailable response. self.getPage("/") self.assertStatus(503) # 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.quickstart(self.server_class) cherrypy.engine.start(blocking=False) self.assertEqual(cherrypy.engine.state, 1) if self.server_class: host = cherrypy.server.socket_host port = cherrypy.server.socket_port 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 engine stop cherrypy.engine.stop() self.assertEqual(cherrypy.engine.state, 0) # Verify that the on_stop_engine function was called self.assertEqual(db_connection.running, False) self.assertEqual(len(db_connection.threads), 0) if not self.server_class: # Once the engine has stopped, we should get a 503 # error again. (If we were running an HTTP server, # then the connection should not even be processed). self.getPage("/") self.assertStatus(503) # Block the main thread now and verify that stop() works. def stoptest(): self.getPage("/") self.assertBody("Hello World") cherrypy.engine.stop() cherrypy.engine.start_with_callback(stoptest) self.assertEqual(cherrypy.engine.state, 0) cherrypy.server.stop() def test_1_Restart(self): cherrypy.server.start() cherrypy.engine.start(blocking=False) # 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.engine.restart() self.assertEqual(cherrypy.engine.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.engine.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.engine.stop() self.assertEqual(cherrypy.engine.state, 0) self.assertEqual(db_connection.running, False) self.assertEqual(len(db_connection.threads), 0) cherrypy.server.stop() def test_2_KeyboardInterrupt(self): if self.server_class: # Raise a keyboard interrupt in the HTTP server's main thread. # We must start the server in this, the main thread cherrypy.engine.start(blocking=False) cherrypy.server.start() self.persistent = True try: # Make the first request and assert there's no "Connection: close". self.getPage("/") self.assertStatus('200 OK') self.assertBody("Hello World") self.assertNoHeader("Connection") cherrypy.server.httpservers.keys()[0].interrupt = KeyboardInterrupt while cherrypy.engine.state != 0: time.sleep(0.1) self.assertEqual(db_connection.running, False) self.assertEqual(len(db_connection.threads), 0) self.assertEqual(cherrypy.engine.state, 0) finally: self.persistent = False # Raise a keyboard interrupt in a page handler; on multithreaded # servers, this should occur in one of the worker threads. # This should raise a BadStatusLine error, since the worker # thread will just die without writing a response. cherrypy.engine.start(blocking=False) cherrypy.server.start() try: self.getPage("/ctrlc") except BadStatusLine: pass else: print self.body self.fail("AssertionError: BadStatusLine not raised") while cherrypy.engine.state != 0: time.sleep(0.1) self.assertEqual(db_connection.running, False) self.assertEqual(len(db_connection.threads), 0) def test_3_Deadlocks(self): cherrypy.engine.start(blocking=False) cherrypy.server.start() try: self.assertNotEqual(cherrypy.engine.monitor_thread, None) # Request a "normal" page. self.assertEqual(cherrypy.engine.servings, []) self.getPage("/") self.assertBody("Hello World") # request.close is called async. while cherrypy.engine.servings: time.sleep(0.01) # Request a page that explicitly checks itself for deadlock. # The deadlock_timeout should be 2 secs. self.getPage("/block_explicit") self.assertBody("broken!") # Request a page that implicitly breaks deadlock. # If we deadlock, we want to touch as little code as possible, # so we won't even call handle_error, just bail ASAP. self.getPage("/block_implicit") self.assertStatus(500) self.assertInBody("raise cherrypy.TimeoutError()") finally: cherrypy.engine.stop() cherrypy.server.stop() def test_4_Autoreload(self): if not self.server_class: print "skipped (no server) ", return # Start the demo script in a new process demoscript = os.path.join(os.getcwd(), os.path.dirname(__file__), "test_states_demo.py") host = cherrypy.server.socket_host port = cherrypy.server.socket_port cherrypy._cpserver.wait_for_free_port(host, port) args = [sys.executable, demoscript, host, str(port)] if self.scheme == "https": args.append('-ssl') pid = os.spawnl(os.P_NOWAIT, sys.executable, *args) pid = str(pid) cherrypy._cpserver.wait_for_occupied_port(host, port) try: self.getPage("/pid") assert self.body.isdigit(), self.body pid = self.body # Give the autoreloader time to cache the file time. time.sleep(2) # Touch the file f = open(demoscript, 'ab') f.write(" ") f.close() # Give the autoreloader time to re-exec the process time.sleep(2) cherrypy._cpserver.wait_for_occupied_port(host, port) self.getPage("/pid") assert self.body.isdigit(), self.body self.assertNotEqual(self.body, pid) pid = self.body finally: # Shut down the spawned process self.getPage("/stop") try: try: # Mac, UNIX print os.wait() except AttributeError: # Windows print os.waitpid(int(pid), 0) except OSError, x: if x.args != (10, 'No child processes'): raise db_connection = None def run(server, conf): helper.setConfig(conf) ServerStateTests.server_class = server suite = helper.CPTestLoader.loadTestsFromTestCase(ServerStateTests) try: global db_connection db_connection = Dependency() cherrypy.engine.on_start_engine_list.append(db_connection.start) cherrypy.engine.on_stop_engine_list.append(db_connection.stop) cherrypy.engine.on_start_thread_list.append(db_connection.startthread) cherrypy.engine.on_stop_thread_list.append(db_connection.stopthread) try: import pyconquer except ImportError: helper.CPTestRunner.run(suite) else: tr = pyconquer.Logger("cherrypy") tr.out = open(os.path.join(os.path.dirname(__file__), "state.log"), "wb") try: tr.start() helper.CPTestRunner.run(suite) finally: tr.stop() tr.out.close() finally: cherrypy.server.stop() cherrypy.engine.stop() def run_all(host, port, ssl=False): conf = {'server.socket_host': host, 'server.socket_port': port, 'server.thread_pool': 10, 'environment': "test_suite", } if host: ServerStateTests.HOST = host if port: ServerStateTests.PORT = port if ssl: localDir = os.path.dirname(__file__) serverpem = os.path.join(os.getcwd(), localDir, 'test.pem') conf['server.ssl_certificate'] = serverpem conf['server.ssl_private_key'] = serverpem ServerStateTests.scheme = "https" ServerStateTests.HTTP_CONN = httplib.HTTPSConnection def _run(server): print print "Testing %s on %s:%s..." % (server, host, port) run(server, conf) _run("cherrypy._cpwsgi.CPWSGIServer") if __name__ == "__main__": import sys host = '127.0.0.1' port = 8000 ssl = False argv = sys.argv[1:] if argv: help_args = [prefix + atom for atom in ("?", "h", "help") for prefix in ("", "-", "--", "\\")] for arg in argv: if arg in help_args: print print "test_states.py -? -> this help page" print "test_states.py [-host=h] [-port=p] -> run the tests on h:p" print "test_states.py -ssl [-host=h] [-port=p] -> run the tests using SSL on h:p" sys.exit(0) if arg == "-ssl": ssl = True elif arg.startswith("-host="): host = arg[6:].strip("\"'") elif arg.startswith("-port="): port = int(arg[6:].strip()) run_all(host, port, ssl)