Package cherrypy :: Package test :: Module test_states
[hide private]
[frames] | no frames]

Source Code for Module cherrypy.test.test_states

  1  import httplib 
  2  from httplib import BadStatusLine 
  3   
  4  import os 
  5  import sys 
  6  import threading 
  7  import time 
  8   
  9  from cherrypy.test import test 
 10  test.prefer_parent_path() 
 11   
 12  import cherrypy 
 13  engine = cherrypy.engine 
 14  thisdir = os.path.join(os.getcwd(), os.path.dirname(__file__)) 
 15   
 16   
17 -class Root:
18 - def index(self):
19 return "Hello World"
20 index.exposed = True 21
22 - def ctrlc(self):
23 raise KeyboardInterrupt()
24 ctrlc.exposed = True 25
26 - def graceful(self):
27 engine.graceful() 28 return "app was (gracefully) restarted succesfully"
29 graceful.exposed = True 30
31 - def block_explicit(self):
32 while True: 33 if cherrypy.response.timed_out: 34 cherrypy.response.timed_out = False 35 return "broken!" 36 time.sleep(0.01)
37 block_explicit.exposed = True 38
39 - def block_implicit(self):
40 time.sleep(0.5) 41 return "response.timeout = %s" % cherrypy.response.timeout
42 block_implicit.exposed = True
43 44 cherrypy.tree.mount(Root()) 45 cherrypy.config.update({ 46 'environment': 'test_suite', 47 'engine.deadlock_poll_freq': 0.1, 48 }) 49
50 -class Dependency:
51
52 - def __init__(self, bus):
53 self.bus = bus 54 self.running = False 55 self.startcount = 0 56 self.gracecount = 0 57 self.threads = {}
58
59 - def subscribe(self):
60 self.bus.subscribe('start', self.start) 61 self.bus.subscribe('stop', self.stop) 62 self.bus.subscribe('graceful', self.graceful) 63 self.bus.subscribe('start_thread', self.startthread) 64 self.bus.subscribe('stop_thread', self.stopthread)
65
66 - def start(self):
67 self.running = True 68 self.startcount += 1
69
70 - def stop(self):
71 self.running = False
72
73 - def graceful(self):
74 self.gracecount += 1
75
76 - def startthread(self, thread_id):
77 self.threads[thread_id] = None
78
79 - def stopthread(self, thread_id):
80 del self.threads[thread_id]
81 82 db_connection = Dependency(engine) 83 db_connection.subscribe() 84 85 86 87 # ------------ Enough helpers. Time for real live test cases. ------------ # 88 89 90 from cherrypy.test import helper 91
92 -class ServerStateTests(helper.CPWebCase):
93
94 - def test_0_NormalStateFlow(self):
95 if not self.server_class: 96 # Without having called "engine.start()", we should 97 # get a 503 Service Unavailable response. 98 self.getPage("/") 99 self.assertStatus(503) 100 101 # And our db_connection should not be running 102 self.assertEqual(db_connection.running, False) 103 self.assertEqual(db_connection.startcount, 0) 104 self.assertEqual(len(db_connection.threads), 0) 105 106 # Test server start 107 engine.start() 108 self.assertEqual(engine.state, engine.states.STARTED) 109 110 if self.server_class: 111 host = cherrypy.server.socket_host 112 port = cherrypy.server.socket_port 113 self.assertRaises(IOError, cherrypy._cpserver.check_port, host, port) 114 115 # The db_connection should be running now 116 self.assertEqual(db_connection.running, True) 117 self.assertEqual(db_connection.startcount, 1) 118 self.assertEqual(len(db_connection.threads), 0) 119 120 self.getPage("/") 121 self.assertBody("Hello World") 122 self.assertEqual(len(db_connection.threads), 1) 123 124 # Test engine stop. This will also stop the HTTP server. 125 engine.stop() 126 self.assertEqual(engine.state, engine.states.STOPPED) 127 128 # Verify that our custom stop function was called 129 self.assertEqual(db_connection.running, False) 130 self.assertEqual(len(db_connection.threads), 0) 131 132 if not self.server_class: 133 # Once the engine has stopped, we should get a 503 134 # error again. (If we were running an HTTP server, 135 # then the connection should not even be processed). 136 self.getPage("/") 137 self.assertStatus(503) 138 139 # Block the main thread now and verify that exit() works. 140 def exittest(): 141 self.getPage("/") 142 self.assertBody("Hello World") 143 engine.exit()
144 cherrypy.server.start() 145 engine.start_with_callback(exittest) 146 engine.block() 147 self.assertEqual(engine.state, engine.states.EXITING)
148
149 - def test_1_Restart(self):
150 cherrypy.server.start() 151 engine.start() 152 153 # The db_connection should be running now 154 self.assertEqual(db_connection.running, True) 155 grace = db_connection.gracecount 156 157 self.getPage("/") 158 self.assertBody("Hello World") 159 self.assertEqual(len(db_connection.threads), 1) 160 161 # Test server restart from this thread 162 engine.graceful() 163 self.assertEqual(engine.state, engine.states.STARTED) 164 self.getPage("/") 165 self.assertBody("Hello World") 166 self.assertEqual(db_connection.running, True) 167 self.assertEqual(db_connection.gracecount, grace + 1) 168 self.assertEqual(len(db_connection.threads), 1) 169 170 # Test server restart from inside a page handler 171 self.getPage("/graceful") 172 self.assertEqual(engine.state, engine.states.STARTED) 173 self.assertBody("app was (gracefully) restarted succesfully") 174 self.assertEqual(db_connection.running, True) 175 self.assertEqual(db_connection.gracecount, grace + 2) 176 # Since we are requesting synchronously, is only one thread used? 177 # Note that the "/graceful" request has been flushed. 178 self.assertEqual(len(db_connection.threads), 0) 179 180 engine.stop() 181 self.assertEqual(engine.state, engine.states.STOPPED) 182 self.assertEqual(db_connection.running, False) 183 self.assertEqual(len(db_connection.threads), 0)
184
185 - def test_2_KeyboardInterrupt(self):
186 if self.server_class: 187 188 # Raise a keyboard interrupt in the HTTP server's main thread. 189 # We must start the server in this, the main thread 190 engine.start() 191 cherrypy.server.start() 192 193 self.persistent = True 194 try: 195 # Make the first request and assert there's no "Connection: close". 196 self.getPage("/") 197 self.assertStatus('200 OK') 198 self.assertBody("Hello World") 199 self.assertNoHeader("Connection") 200 201 cherrypy.server.httpserver.interrupt = KeyboardInterrupt 202 engine.block() 203 204 self.assertEqual(db_connection.running, False) 205 self.assertEqual(len(db_connection.threads), 0) 206 self.assertEqual(engine.state, engine.states.EXITING) 207 finally: 208 self.persistent = False 209 210 # Raise a keyboard interrupt in a page handler; on multithreaded 211 # servers, this should occur in one of the worker threads. 212 # This should raise a BadStatusLine error, since the worker 213 # thread will just die without writing a response. 214 engine.start() 215 cherrypy.server.start() 216 217 try: 218 self.getPage("/ctrlc") 219 except BadStatusLine: 220 pass 221 else: 222 print self.body 223 self.fail("AssertionError: BadStatusLine not raised") 224 225 engine.block() 226 self.assertEqual(db_connection.running, False) 227 self.assertEqual(len(db_connection.threads), 0)
228
229 - def test_3_Deadlocks(self):
230 cherrypy.config.update({'response.timeout': 0.2}) 231 232 engine.start() 233 cherrypy.server.start() 234 try: 235 self.assertNotEqual(engine.timeout_monitor.thread, None) 236 237 # Request a "normal" page. 238 self.assertEqual(engine.timeout_monitor.servings, []) 239 self.getPage("/") 240 self.assertBody("Hello World") 241 # request.close is called async. 242 while engine.timeout_monitor.servings: 243 print ".", 244 time.sleep(0.01) 245 246 # Request a page that explicitly checks itself for deadlock. 247 # The deadlock_timeout should be 2 secs. 248 self.getPage("/block_explicit") 249 self.assertBody("broken!") 250 251 # Request a page that implicitly breaks deadlock. 252 # If we deadlock, we want to touch as little code as possible, 253 # so we won't even call handle_error, just bail ASAP. 254 self.getPage("/block_implicit") 255 self.assertStatus(500) 256 self.assertInBody("raise cherrypy.TimeoutError()") 257 finally: 258 engine.exit()
259
260 - def test_4_Autoreload(self):
261 if not self.server_class: 262 print "skipped (no server) ", 263 return 264 265 # Start the demo script in a new process 266 p = helper.CPProcess(ssl=(self.scheme.lower()=='https')) 267 p.write_conf() 268 p.start(imports='cherrypy.test.test_states_demo') 269 try: 270 self.getPage("/start") 271 start = float(self.body) 272 273 # Give the autoreloader time to cache the file time. 274 time.sleep(2) 275 276 # Touch the file 277 os.utime(os.path.join(thisdir, "test_states_demo.py"), None) 278 279 # Give the autoreloader time to re-exec the process 280 time.sleep(2) 281 cherrypy._cpserver.wait_for_occupied_port(host, port) 282 283 self.getPage("/start") 284 self.assert_(float(self.body) > start) 285 finally: 286 # Shut down the spawned process 287 self.getPage("/exit") 288 p.join()
289
290 - def test_5_Start_Error(self):
291 if not self.server_class: 292 print "skipped (no server) ", 293 return 294 295 # If a process errors during start, it should stop the engine 296 # and exit with a non-zero exit code. 297 p = helper.CPProcess(ssl=(self.scheme.lower()=='https'), 298 wait=True) 299 p.write_conf(extra="starterror: True") 300 p.start(imports='cherrypy.test.test_states_demo') 301 if p.exit_code == 0: 302 self.fail("Process failed to return nonzero exit code.")
303 304
305 -class PluginTests(helper.CPWebCase):
306
307 - def test_daemonize(self):
308 if not self.server_class: 309 print "skipped (no server) ", 310 return 311 if os.name not in ['posix']: 312 print "skipped (not on posix) ", 313 return 314 315 # Spawn the process and wait, when this returns, the original process 316 # is finished. If it daemonized properly, we should still be able 317 # to access pages. 318 p = helper.CPProcess(ssl=(self.scheme.lower()=='https'), 319 wait=True, daemonize=True) 320 p.write_conf() 321 p.start(imports='cherrypy.test.test_states_demo') 322 try: 323 # Just get the pid of the daemonization process. 324 self.getPage("/pid") 325 self.assertStatus(200) 326 page_pid = int(self.body) 327 self.assertEqual(page_pid, p.get_pid()) 328 finally: 329 # Shut down the spawned process 330 self.getPage("/exit") 331 p.join() 332 333 # Wait until here to test the exit code because we want to ensure 334 # that we wait for the daemon to finish running before we fail. 335 if p.exit_code != 0: 336 self.fail("Daemonized parent process failed to exit cleanly.")
337 338
339 -class SignalHandlingTests(helper.CPWebCase):
340
341 - def test_SIGHUP_tty(self):
342 # When not daemonized, SIGHUP should shut down the server. 343 if not self.server_class: 344 print "skipped (no server) ", 345 return 346 347 try: 348 from signal import SIGHUP 349 except ImportError: 350 print "skipped (no SIGHUP) ", 351 return 352 353 # Spawn the process. 354 p = helper.CPProcess(ssl=(self.scheme.lower()=='https')) 355 p.write_conf() 356 p.start(imports='cherrypy.test.test_states_demo') 357 # Send a SIGHUP 358 os.kill(p.get_pid(), SIGHUP) 359 # This might hang if things aren't working right, but meh. 360 p.join()
361
362 - def test_SIGHUP_daemonized(self):
363 # When daemonized, SIGHUP should restart the server. 364 if not self.server_class: 365 print "skipped (no server) ", 366 return 367 368 try: 369 from signal import SIGHUP 370 except ImportError: 371 print "skipped (no SIGHUP) ", 372 return 373 374 if os.name not in ['posix']: 375 print "skipped (not on posix) ", 376 return 377 378 # Spawn the process and wait, when this returns, the original process 379 # is finished. If it daemonized properly, we should still be able 380 # to access pages. 381 p = helper.CPProcess(ssl=(self.scheme.lower()=='https'), 382 wait=True, daemonize=True) 383 p.write_conf() 384 p.start(imports='cherrypy.test.test_states_demo') 385 386 pid = p.get_pid() 387 try: 388 # Send a SIGHUP 389 os.kill(pid, SIGHUP) 390 # Give the server some time to restart 391 time.sleep(2) 392 self.getPage("/pid") 393 self.assertStatus(200) 394 new_pid = int(self.body) 395 self.assertNotEqual(new_pid, pid) 396 finally: 397 # Shut down the spawned process 398 self.getPage("/exit") 399 p.join()
400
401 - def test_SIGTERM(self):
402 # SIGTERM should shut down the server whether daemonized or not. 403 if not self.server_class: 404 print "skipped (no server) ", 405 return 406 407 try: 408 from signal import SIGTERM 409 except ImportError: 410 print "skipped (no SIGTERM) ", 411 return 412 413 try: 414 from os import kill 415 except ImportError: 416 print "skipped (no os.kill) ", 417 return 418 419 # Spawn a normal, undaemonized process. 420 p = helper.CPProcess(ssl=(self.scheme.lower()=='https')) 421 p.write_conf() 422 p.start(imports='cherrypy.test.test_states_demo') 423 # Send a SIGTERM 424 os.kill(p.get_pid(), SIGTERM) 425 # This might hang if things aren't working right, but meh. 426 p.join() 427 428 if os.name in ['posix']: 429 # Spawn a daemonized process and test again. 430 p = helper.CPProcess(ssl=(self.scheme.lower()=='https'), 431 wait=True, daemonize=True) 432 p.write_conf() 433 p.start(imports='cherrypy.test.test_states_demo') 434 # Send a SIGTERM 435 os.kill(p.get_pid(), SIGTERM) 436 # This might hang if things aren't working right, but meh. 437 p.join()
438
440 if not self.server_class: 441 print "skipped (no server) ", 442 return 443 444 try: 445 from signal import SIGTERM 446 except ImportError: 447 print "skipped (no SIGTERM) ", 448 return 449 450 try: 451 from os import kill 452 except ImportError: 453 print "skipped (no os.kill) ", 454 return 455 456 # Spawn a normal, undaemonized process. 457 p = helper.CPProcess(ssl=(self.scheme.lower()=='https')) 458 p.write_conf(extra="unsubsig: True") 459 p.start(imports='cherrypy.test.test_states_demo') 460 # Send a SIGTERM 461 os.kill(p.get_pid(), SIGTERM) 462 # This might hang if things aren't working right, but meh. 463 p.join() 464 465 # Assert the old handler ran. 466 target_line = open(p.error_log, 'rb').readlines()[-10] 467 if not "I am an old SIGTERM handler." in target_line: 468 self.fail("Old SIGTERM handler did not run.\n%r" % target_line)
469 470 471 cases = [v for v in globals().values() 472 if isinstance(v, type) and issubclass(v, helper.CPWebCase)] 473
474 -def run(server, conf):
475 helper.setConfig(conf) 476 for tc in cases: 477 tc.server_class = server 478 suites = [helper.CPTestLoader.loadTestsFromTestCase(tc) for tc in 479 (ServerStateTests, PluginTests, SignalHandlingTests)] 480 try: 481 try: 482 import pyconquer 483 except ImportError: 484 for suite in suites: 485 helper.CPTestRunner.run(suite) 486 else: 487 tr = pyconquer.Logger("cherrypy") 488 tr.out = open(os.path.join(os.path.dirname(__file__), "test_states_conquer.log"), "wb") 489 try: 490 tr.start() 491 for suite in suites: 492 helper.CPTestRunner.run(suite) 493 finally: 494 tr.stop() 495 tr.out.close() 496 finally: 497 engine.exit()
498 499
500 -def run_all(host, port, ssl=False):
501 conf = {'server.socket_host': host, 502 'server.socket_port': port, 503 'server.thread_pool': 10, 504 'environment': "test_suite", 505 } 506 507 if host: 508 for tc in cases: 509 tc.HOST = host 510 511 if port: 512 for tc in cases: 513 tc.PORT = port 514 515 if ssl: 516 localDir = os.path.dirname(__file__) 517 serverpem = os.path.join(os.getcwd(), localDir, 'test.pem') 518 conf['server.ssl_certificate'] = serverpem 519 conf['server.ssl_private_key'] = serverpem 520 for tc in cases: 521 tc.scheme = "https" 522 tc.HTTP_CONN = httplib.HTTPSConnection 523 524 def _run(server): 525 print 526 print "Testing %s on %s:%s..." % (server, host, port) 527 run(server, conf)
528 _run("cherrypy._cpwsgi.CPWSGIServer") 529 530 if __name__ == "__main__": 531 import sys 532 533 host = '127.0.0.1' 534 port = 8000 535 ssl = False 536 537 argv = sys.argv[1:] 538 if argv: 539 help_args = [prefix + atom for atom in ("?", "h", "help") 540 for prefix in ("", "-", "--", "\\")] 541 542 for arg in argv: 543 if arg in help_args: 544 print 545 print "test_states.py -? -> this help page" 546 print "test_states.py [-host=h] [-port=p] -> run the tests on h:p" 547 print "test_states.py -ssl [-host=h] [-port=p] -> run the tests using SSL on h:p" 548 sys.exit(0) 549 550 if arg == "-ssl": 551 ssl = True 552 elif arg.startswith("-host="): 553 host = arg[6:].strip("\"'") 554 elif arg.startswith("-port="): 555 port = int(arg[6:].strip()) 556 557 run_all(host, port, ssl) 558