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

Source Code for Module cherrypy.test.test_states

  1  from cherrypy._cpcompat import BadStatusLine, ntob 
  2  import os 
  3  import sys 
  4  import threading 
  5  import time 
  6   
  7  import cherrypy 
  8  engine = cherrypy.engine 
  9  thisdir = os.path.join(os.getcwd(), os.path.dirname(__file__)) 
 10   
 11   
12 -class Dependency:
13
14 - def __init__(self, bus):
15 self.bus = bus 16 self.running = False 17 self.startcount = 0 18 self.gracecount = 0 19 self.threads = {}
20
21 - def subscribe(self):
22 self.bus.subscribe('start', self.start) 23 self.bus.subscribe('stop', self.stop) 24 self.bus.subscribe('graceful', self.graceful) 25 self.bus.subscribe('start_thread', self.startthread) 26 self.bus.subscribe('stop_thread', self.stopthread)
27
28 - def start(self):
29 self.running = True 30 self.startcount += 1
31
32 - def stop(self):
33 self.running = False
34
35 - def graceful(self):
36 self.gracecount += 1
37
38 - def startthread(self, thread_id):
39 self.threads[thread_id] = None
40
41 - def stopthread(self, thread_id):
42 del self.threads[thread_id]
43 44 db_connection = Dependency(engine) 45
46 -def setup_server():
47 class Root: 48 def index(self): 49 return "Hello World"
50 index.exposed = True 51 52 def ctrlc(self): 53 raise KeyboardInterrupt() 54 ctrlc.exposed = True 55 56 def graceful(self): 57 engine.graceful() 58 return "app was (gracefully) restarted succesfully" 59 graceful.exposed = True 60 61 def block_explicit(self): 62 while True: 63 if cherrypy.response.timed_out: 64 cherrypy.response.timed_out = False 65 return "broken!" 66 time.sleep(0.01) 67 block_explicit.exposed = True 68 69 def block_implicit(self): 70 time.sleep(0.5) 71 return "response.timeout = %s" % cherrypy.response.timeout 72 block_implicit.exposed = True 73 74 cherrypy.tree.mount(Root()) 75 cherrypy.config.update({ 76 'environment': 'test_suite', 77 'engine.deadlock_poll_freq': 0.1, 78 }) 79 80 db_connection.subscribe() 81 82 83 84 # ------------ Enough helpers. Time for real live test cases. ------------ # 85 86 87 from cherrypy.test import helper 88
89 -class ServerStateTests(helper.CPWebCase):
90 setup_server = staticmethod(setup_server) 91
92 - def setUp(self):
93 cherrypy.server.socket_timeout = 0.1 94 self.do_gc_test = False
95
96 - def test_0_NormalStateFlow(self):
97 engine.stop() 98 # Our db_connection should not be running 99 self.assertEqual(db_connection.running, False) 100 self.assertEqual(db_connection.startcount, 1) 101 self.assertEqual(len(db_connection.threads), 0) 102 103 # Test server start 104 engine.start() 105 self.assertEqual(engine.state, engine.states.STARTED) 106 107 host = cherrypy.server.socket_host 108 port = cherrypy.server.socket_port 109 self.assertRaises(IOError, cherrypy._cpserver.check_port, host, port) 110 111 # The db_connection should be running now 112 self.assertEqual(db_connection.running, True) 113 self.assertEqual(db_connection.startcount, 2) 114 self.assertEqual(len(db_connection.threads), 0) 115 116 self.getPage("/") 117 self.assertBody("Hello World") 118 self.assertEqual(len(db_connection.threads), 1) 119 120 # Test engine stop. This will also stop the HTTP server. 121 engine.stop() 122 self.assertEqual(engine.state, engine.states.STOPPED) 123 124 # Verify that our custom stop function was called 125 self.assertEqual(db_connection.running, False) 126 self.assertEqual(len(db_connection.threads), 0) 127 128 # Block the main thread now and verify that exit() works. 129 def exittest(): 130 self.getPage("/") 131 self.assertBody("Hello World") 132 engine.exit()
133 cherrypy.server.start() 134 engine.start_with_callback(exittest) 135 engine.block() 136 self.assertEqual(engine.state, engine.states.EXITING)
137
138 - def test_1_Restart(self):
139 cherrypy.server.start() 140 engine.start() 141 142 # The db_connection should be running now 143 self.assertEqual(db_connection.running, True) 144 grace = db_connection.gracecount 145 146 self.getPage("/") 147 self.assertBody("Hello World") 148 self.assertEqual(len(db_connection.threads), 1) 149 150 # Test server restart from this thread 151 engine.graceful() 152 self.assertEqual(engine.state, engine.states.STARTED) 153 self.getPage("/") 154 self.assertBody("Hello World") 155 self.assertEqual(db_connection.running, True) 156 self.assertEqual(db_connection.gracecount, grace + 1) 157 self.assertEqual(len(db_connection.threads), 1) 158 159 # Test server restart from inside a page handler 160 self.getPage("/graceful") 161 self.assertEqual(engine.state, engine.states.STARTED) 162 self.assertBody("app was (gracefully) restarted succesfully") 163 self.assertEqual(db_connection.running, True) 164 self.assertEqual(db_connection.gracecount, grace + 2) 165 # Since we are requesting synchronously, is only one thread used? 166 # Note that the "/graceful" request has been flushed. 167 self.assertEqual(len(db_connection.threads), 0) 168 169 engine.stop() 170 self.assertEqual(engine.state, engine.states.STOPPED) 171 self.assertEqual(db_connection.running, False) 172 self.assertEqual(len(db_connection.threads), 0)
173
174 - def test_2_KeyboardInterrupt(self):
175 # Raise a keyboard interrupt in the HTTP server's main thread. 176 # We must start the server in this, the main thread 177 engine.start() 178 cherrypy.server.start() 179 180 self.persistent = True 181 try: 182 # Make the first request and assert there's no "Connection: close". 183 self.getPage("/") 184 self.assertStatus('200 OK') 185 self.assertBody("Hello World") 186 self.assertNoHeader("Connection") 187 188 cherrypy.server.httpserver.interrupt = KeyboardInterrupt 189 engine.block() 190 191 self.assertEqual(db_connection.running, False) 192 self.assertEqual(len(db_connection.threads), 0) 193 self.assertEqual(engine.state, engine.states.EXITING) 194 finally: 195 self.persistent = False 196 197 # Raise a keyboard interrupt in a page handler; on multithreaded 198 # servers, this should occur in one of the worker threads. 199 # This should raise a BadStatusLine error, since the worker 200 # thread will just die without writing a response. 201 engine.start() 202 cherrypy.server.start() 203 204 try: 205 self.getPage("/ctrlc") 206 except BadStatusLine: 207 pass 208 else: 209 print(self.body) 210 self.fail("AssertionError: BadStatusLine not raised") 211 212 engine.block() 213 self.assertEqual(db_connection.running, False) 214 self.assertEqual(len(db_connection.threads), 0)
215
216 - def test_3_Deadlocks(self):
217 cherrypy.config.update({'response.timeout': 0.2}) 218 219 engine.start() 220 cherrypy.server.start() 221 try: 222 self.assertNotEqual(engine.timeout_monitor.thread, None) 223 224 # Request a "normal" page. 225 self.assertEqual(engine.timeout_monitor.servings, []) 226 self.getPage("/") 227 self.assertBody("Hello World") 228 # request.close is called async. 229 while engine.timeout_monitor.servings: 230 sys.stdout.write(".") 231 time.sleep(0.01) 232 233 # Request a page that explicitly checks itself for deadlock. 234 # The deadlock_timeout should be 2 secs. 235 self.getPage("/block_explicit") 236 self.assertBody("broken!") 237 238 # Request a page that implicitly breaks deadlock. 239 # If we deadlock, we want to touch as little code as possible, 240 # so we won't even call handle_error, just bail ASAP. 241 self.getPage("/block_implicit") 242 self.assertStatus(500) 243 self.assertInBody("raise cherrypy.TimeoutError()") 244 finally: 245 engine.exit()
246
247 - def test_4_Autoreload(self):
248 # Start the demo script in a new process 249 p = helper.CPProcess(ssl=(self.scheme.lower()=='https')) 250 p.write_conf( 251 extra='test_case_name: "test_4_Autoreload"') 252 p.start(imports='cherrypy.test._test_states_demo') 253 try: 254 self.getPage("/start") 255 start = float(self.body) 256 257 # Give the autoreloader time to cache the file time. 258 time.sleep(2) 259 260 # Touch the file 261 os.utime(os.path.join(thisdir, "_test_states_demo.py"), None) 262 263 # Give the autoreloader time to re-exec the process 264 time.sleep(2) 265 host = cherrypy.server.socket_host 266 port = cherrypy.server.socket_port 267 cherrypy._cpserver.wait_for_occupied_port(host, port) 268 269 self.getPage("/start") 270 if not (float(self.body) > start): 271 raise AssertionError("start time %s not greater than %s" % 272 (float(self.body), start)) 273 finally: 274 # Shut down the spawned process 275 self.getPage("/exit") 276 p.join()
277
278 - def test_5_Start_Error(self):
279 # If a process errors during start, it should stop the engine 280 # and exit with a non-zero exit code. 281 p = helper.CPProcess(ssl=(self.scheme.lower()=='https'), 282 wait=True) 283 p.write_conf( 284 extra="""starterror: True 285 test_case_name: "test_5_Start_Error" 286 """ 287 ) 288 p.start(imports='cherrypy.test._test_states_demo') 289 if p.exit_code == 0: 290 self.fail("Process failed to return nonzero exit code.")
291 292
293 -class PluginTests(helper.CPWebCase):
294 - def test_daemonize(self):
295 if os.name not in ['posix']: 296 return self.skip("skipped (not on posix) ") 297 self.HOST = '127.0.0.1' 298 self.PORT = 8081 299 # Spawn the process and wait, when this returns, the original process 300 # is finished. If it daemonized properly, we should still be able 301 # to access pages. 302 p = helper.CPProcess(ssl=(self.scheme.lower()=='https'), 303 wait=True, daemonize=True, 304 socket_host='127.0.0.1', 305 socket_port=8081) 306 p.write_conf( 307 extra='test_case_name: "test_daemonize"') 308 p.start(imports='cherrypy.test._test_states_demo') 309 try: 310 # Just get the pid of the daemonization process. 311 self.getPage("/pid") 312 self.assertStatus(200) 313 page_pid = int(self.body) 314 self.assertEqual(page_pid, p.get_pid()) 315 finally: 316 # Shut down the spawned process 317 self.getPage("/exit") 318 p.join() 319 320 # Wait until here to test the exit code because we want to ensure 321 # that we wait for the daemon to finish running before we fail. 322 if p.exit_code != 0: 323 self.fail("Daemonized parent process failed to exit cleanly.")
324 325
326 -class SignalHandlingTests(helper.CPWebCase):
327 - def test_SIGHUP_tty(self):
328 # When not daemonized, SIGHUP should shut down the server. 329 try: 330 from signal import SIGHUP 331 except ImportError: 332 return self.skip("skipped (no SIGHUP) ") 333 334 # Spawn the process. 335 p = helper.CPProcess(ssl=(self.scheme.lower()=='https')) 336 p.write_conf( 337 extra='test_case_name: "test_SIGHUP_tty"') 338 p.start(imports='cherrypy.test._test_states_demo') 339 # Send a SIGHUP 340 os.kill(p.get_pid(), SIGHUP) 341 # This might hang if things aren't working right, but meh. 342 p.join()
343
344 - def test_SIGHUP_daemonized(self):
345 # When daemonized, SIGHUP should restart the server. 346 try: 347 from signal import SIGHUP 348 except ImportError: 349 return self.skip("skipped (no SIGHUP) ") 350 351 if os.name not in ['posix']: 352 return self.skip("skipped (not on posix) ") 353 354 # Spawn the process and wait, when this returns, the original process 355 # is finished. If it daemonized properly, we should still be able 356 # to access pages. 357 p = helper.CPProcess(ssl=(self.scheme.lower()=='https'), 358 wait=True, daemonize=True) 359 p.write_conf( 360 extra='test_case_name: "test_SIGHUP_daemonized"') 361 p.start(imports='cherrypy.test._test_states_demo') 362 363 pid = p.get_pid() 364 try: 365 # Send a SIGHUP 366 os.kill(pid, SIGHUP) 367 # Give the server some time to restart 368 time.sleep(2) 369 self.getPage("/pid") 370 self.assertStatus(200) 371 new_pid = int(self.body) 372 self.assertNotEqual(new_pid, pid) 373 finally: 374 # Shut down the spawned process 375 self.getPage("/exit") 376 p.join()
377
378 - def test_SIGTERM(self):
379 # SIGTERM should shut down the server whether daemonized or not. 380 try: 381 from signal import SIGTERM 382 except ImportError: 383 return self.skip("skipped (no SIGTERM) ") 384 385 try: 386 from os import kill 387 except ImportError: 388 return self.skip("skipped (no os.kill) ") 389 390 # Spawn a normal, undaemonized process. 391 p = helper.CPProcess(ssl=(self.scheme.lower()=='https')) 392 p.write_conf( 393 extra='test_case_name: "test_SIGTERM"') 394 p.start(imports='cherrypy.test._test_states_demo') 395 # Send a SIGTERM 396 os.kill(p.get_pid(), SIGTERM) 397 # This might hang if things aren't working right, but meh. 398 p.join() 399 400 if os.name in ['posix']: 401 # Spawn a daemonized process and test again. 402 p = helper.CPProcess(ssl=(self.scheme.lower()=='https'), 403 wait=True, daemonize=True) 404 p.write_conf( 405 extra='test_case_name: "test_SIGTERM_2"') 406 p.start(imports='cherrypy.test._test_states_demo') 407 # Send a SIGTERM 408 os.kill(p.get_pid(), SIGTERM) 409 # This might hang if things aren't working right, but meh. 410 p.join()
411
413 try: 414 from signal import SIGTERM 415 except ImportError: 416 return self.skip("skipped (no SIGTERM) ") 417 418 try: 419 from os import kill 420 except ImportError: 421 return self.skip("skipped (no os.kill) ") 422 423 # Spawn a normal, undaemonized process. 424 p = helper.CPProcess(ssl=(self.scheme.lower()=='https')) 425 p.write_conf( 426 extra="""unsubsig: True 427 test_case_name: "test_signal_handler_unsubscribe" 428 """) 429 p.start(imports='cherrypy.test._test_states_demo') 430 # Send a SIGTERM 431 os.kill(p.get_pid(), SIGTERM) 432 # This might hang if things aren't working right, but meh. 433 p.join() 434 435 # Assert the old handler ran. 436 target_line = open(p.error_log, 'rb').readlines()[-10] 437 if not ntob("I am an old SIGTERM handler.") in target_line: 438 self.fail("Old SIGTERM handler did not run.\n%r" % target_line)
439