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   
 14   
15 -class Root:
16 - def index(self):
17 return "Hello World"
18 index.exposed = True 19
20 - def ctrlc(self):
21 raise KeyboardInterrupt()
22 ctrlc.exposed = True 23
24 - def restart(self):
25 cherrypy.engine.restart() 26 return "app was restarted succesfully"
27 restart.exposed = True 28
29 - def block_explicit(self):
30 while True: 31 if cherrypy.response.timed_out: 32 cherrypy.response.timed_out = False 33 return "broken!" 34 time.sleep(0.01)
35 block_explicit.exposed = True 36
37 - def block_implicit(self):
38 time.sleep(0.5) 39 return "response.timeout = %s" % cherrypy.response.timeout
40 block_implicit.exposed = True
41 42 cherrypy.tree.mount(Root()) 43 cherrypy.config.update({ 44 'environment': 'test_suite', 45 'engine.deadlock_poll_freq': 0.1, 46 'response.timeout': 0.2, 47 }) 48
49 -class Dependency:
50
51 - def __init__(self):
52 self.running = False 53 self.startcount = 0 54 self.threads = {}
55
56 - def start(self):
57 self.running = True 58 self.startcount += 1
59
60 - def stop(self):
61 self.running = False
62
63 - def startthread(self, thread_id):
64 self.threads[thread_id] = None
65
66 - def stopthread(self, thread_id):
67 del self.threads[thread_id]
68 69 70 from cherrypy.test import helper 71
72 -class ServerStateTests(helper.CPWebCase):
73
74 - def test_0_NormalStateFlow(self):
75 if not self.server_class: 76 # Without having called "cherrypy.engine.start()", we should 77 # get a 503 Service Unavailable response. 78 self.getPage("/") 79 self.assertStatus(503) 80 81 # And our db_connection should not be running 82 self.assertEqual(db_connection.running, False) 83 self.assertEqual(db_connection.startcount, 0) 84 self.assertEqual(len(db_connection.threads), 0) 85 86 # Test server start 87 cherrypy.server.quickstart(self.server_class) 88 cherrypy.engine.start(blocking=False) 89 self.assertEqual(cherrypy.engine.state, 1) 90 91 if self.server_class: 92 host = cherrypy.server.socket_host 93 port = cherrypy.server.socket_port 94 self.assertRaises(IOError, cherrypy._cpserver.check_port, host, port) 95 96 # The db_connection should be running now 97 self.assertEqual(db_connection.running, True) 98 self.assertEqual(db_connection.startcount, 1) 99 self.assertEqual(len(db_connection.threads), 0) 100 101 self.getPage("/") 102 self.assertBody("Hello World") 103 self.assertEqual(len(db_connection.threads), 1) 104 105 # Test engine stop 106 cherrypy.engine.stop() 107 self.assertEqual(cherrypy.engine.state, 0) 108 109 # Verify that the on_stop_engine function was called 110 self.assertEqual(db_connection.running, False) 111 self.assertEqual(len(db_connection.threads), 0) 112 113 if not self.server_class: 114 # Once the engine has stopped, we should get a 503 115 # error again. (If we were running an HTTP server, 116 # then the connection should not even be processed). 117 self.getPage("/") 118 self.assertStatus(503) 119 120 # Block the main thread now and verify that stop() works. 121 def stoptest(): 122 self.getPage("/") 123 self.assertBody("Hello World") 124 cherrypy.engine.stop()
125 cherrypy.engine.start_with_callback(stoptest) 126 self.assertEqual(cherrypy.engine.state, 0) 127 cherrypy.server.stop()
128
129 - def test_1_Restart(self):
130 cherrypy.server.start() 131 cherrypy.engine.start(blocking=False) 132 133 # The db_connection should be running now 134 self.assertEqual(db_connection.running, True) 135 sc = db_connection.startcount 136 137 self.getPage("/") 138 self.assertBody("Hello World") 139 self.assertEqual(len(db_connection.threads), 1) 140 141 # Test server restart from this thread 142 cherrypy.engine.restart() 143 self.assertEqual(cherrypy.engine.state, 1) 144 self.getPage("/") 145 self.assertBody("Hello World") 146 self.assertEqual(db_connection.running, True) 147 self.assertEqual(db_connection.startcount, sc + 1) 148 self.assertEqual(len(db_connection.threads), 1) 149 150 # Test server restart from inside a page handler 151 self.getPage("/restart") 152 self.assertEqual(cherrypy.engine.state, 1) 153 self.assertBody("app was restarted succesfully") 154 self.assertEqual(db_connection.running, True) 155 self.assertEqual(db_connection.startcount, sc + 2) 156 # Since we are requesting synchronously, is only one thread used? 157 # Note that the "/restart" request has been flushed. 158 self.assertEqual(len(db_connection.threads), 0) 159 160 cherrypy.engine.stop() 161 self.assertEqual(cherrypy.engine.state, 0) 162 self.assertEqual(db_connection.running, False) 163 self.assertEqual(len(db_connection.threads), 0) 164 cherrypy.server.stop()
165
166 - def test_2_KeyboardInterrupt(self):
167 if self.server_class: 168 169 # Raise a keyboard interrupt in the HTTP server's main thread. 170 # We must start the server in this, the main thread 171 cherrypy.engine.start(blocking=False) 172 cherrypy.server.start() 173 174 self.persistent = True 175 try: 176 # Make the first request and assert there's no "Connection: close". 177 self.getPage("/") 178 self.assertStatus('200 OK') 179 self.assertBody("Hello World") 180 self.assertNoHeader("Connection") 181 182 cherrypy.server.httpservers.keys()[0].interrupt = KeyboardInterrupt 183 while cherrypy.engine.state != 0: 184 time.sleep(0.1) 185 186 self.assertEqual(db_connection.running, False) 187 self.assertEqual(len(db_connection.threads), 0) 188 self.assertEqual(cherrypy.engine.state, 0) 189 finally: 190 self.persistent = False 191 192 # Raise a keyboard interrupt in a page handler; on multithreaded 193 # servers, this should occur in one of the worker threads. 194 # This should raise a BadStatusLine error, since the worker 195 # thread will just die without writing a response. 196 cherrypy.engine.start(blocking=False) 197 cherrypy.server.start() 198 199 try: 200 self.getPage("/ctrlc") 201 except BadStatusLine: 202 pass 203 else: 204 print self.body 205 self.fail("AssertionError: BadStatusLine not raised") 206 207 while cherrypy.engine.state != 0: 208 time.sleep(0.1) 209 self.assertEqual(db_connection.running, False) 210 self.assertEqual(len(db_connection.threads), 0)
211
212 - def test_3_Deadlocks(self):
213 cherrypy.engine.start(blocking=False) 214 cherrypy.server.start() 215 try: 216 self.assertNotEqual(cherrypy.engine.monitor_thread, None) 217 218 # Request a "normal" page. 219 self.assertEqual(cherrypy.engine.servings, []) 220 self.getPage("/") 221 self.assertBody("Hello World") 222 # request.close is called async. 223 while cherrypy.engine.servings: 224 time.sleep(0.01) 225 226 # Request a page that explicitly checks itself for deadlock. 227 # The deadlock_timeout should be 2 secs. 228 self.getPage("/block_explicit") 229 self.assertBody("broken!") 230 231 # Request a page that implicitly breaks deadlock. 232 # If we deadlock, we want to touch as little code as possible, 233 # so we won't even call handle_error, just bail ASAP. 234 self.getPage("/block_implicit") 235 self.assertStatus(500) 236 self.assertInBody("raise cherrypy.TimeoutError()") 237 finally: 238 cherrypy.engine.stop() 239 cherrypy.server.stop()
240
241 - def test_4_Autoreload(self):
242 if not self.server_class: 243 print "skipped (no server) ", 244 return 245 246 # Start the demo script in a new process 247 demoscript = os.path.join(os.getcwd(), os.path.dirname(__file__), 248 "test_states_demo.py") 249 host = cherrypy.server.socket_host 250 port = cherrypy.server.socket_port 251 cherrypy._cpserver.wait_for_free_port(host, port) 252 253 args = [sys.executable, demoscript, host, str(port)] 254 if self.scheme == "https": 255 args.append('-ssl') 256 pid = os.spawnl(os.P_NOWAIT, sys.executable, *args) 257 pid = str(pid) 258 cherrypy._cpserver.wait_for_occupied_port(host, port) 259 260 try: 261 self.getPage("/pid") 262 assert self.body.isdigit(), self.body 263 pid = self.body 264 265 # Give the autoreloader time to cache the file time. 266 time.sleep(2) 267 268 # Touch the file 269 f = open(demoscript, 'ab') 270 f.write(" ") 271 f.close() 272 273 # Give the autoreloader time to re-exec the process 274 time.sleep(2) 275 cherrypy._cpserver.wait_for_occupied_port(host, port) 276 277 self.getPage("/pid") 278 assert self.body.isdigit(), self.body 279 self.assertNotEqual(self.body, pid) 280 pid = self.body 281 finally: 282 # Shut down the spawned process 283 self.getPage("/stop") 284 285 try: 286 try: 287 # Mac, UNIX 288 print os.wait() 289 except AttributeError: 290 # Windows 291 print os.waitpid(int(pid), 0) 292 except OSError, x: 293 if x.args != (10, 'No child processes'): 294 raise
295 296 db_connection = None 297
298 -def run(server, conf):
299 helper.setConfig(conf) 300 ServerStateTests.server_class = server 301 suite = helper.CPTestLoader.loadTestsFromTestCase(ServerStateTests) 302 try: 303 global db_connection 304 db_connection = Dependency() 305 cherrypy.engine.on_start_engine_list.append(db_connection.start) 306 cherrypy.engine.on_stop_engine_list.append(db_connection.stop) 307 cherrypy.engine.on_start_thread_list.append(db_connection.startthread) 308 cherrypy.engine.on_stop_thread_list.append(db_connection.stopthread) 309 310 try: 311 import pyconquer 312 except ImportError: 313 helper.CPTestRunner.run(suite) 314 else: 315 tr = pyconquer.Logger("cherrypy") 316 tr.out = open(os.path.join(os.path.dirname(__file__), "state.log"), "wb") 317 try: 318 tr.start() 319 helper.CPTestRunner.run(suite) 320 finally: 321 tr.stop() 322 tr.out.close() 323 finally: 324 cherrypy.server.stop() 325 cherrypy.engine.stop()
326 327
328 -def run_all(host, port, ssl=False):
329 conf = {'server.socket_host': host, 330 'server.socket_port': port, 331 'server.thread_pool': 10, 332 'environment': "test_suite", 333 } 334 335 if host: 336 ServerStateTests.HOST = host 337 338 if port: 339 ServerStateTests.PORT = port 340 341 if ssl: 342 localDir = os.path.dirname(__file__) 343 serverpem = os.path.join(os.getcwd(), localDir, 'test.pem') 344 conf['server.ssl_certificate'] = serverpem 345 conf['server.ssl_private_key'] = serverpem 346 ServerStateTests.scheme = "https" 347 ServerStateTests.HTTP_CONN = httplib.HTTPSConnection 348 349 def _run(server): 350 print 351 print "Testing %s on %s:%s..." % (server, host, port) 352 run(server, conf)
353 _run("cherrypy._cpwsgi.CPWSGIServer") 354 355 356 357 if __name__ == "__main__": 358 import sys 359 360 host = '127.0.0.1' 361 port = 8000 362 ssl = False 363 364 argv = sys.argv[1:] 365 if argv: 366 help_args = [prefix + atom for atom in ("?", "h", "help") 367 for prefix in ("", "-", "--", "\\")] 368 369 for arg in argv: 370 if arg in help_args: 371 print 372 print "test_states.py -? -> this help page" 373 print "test_states.py [-host=h] [-port=p] -> run the tests on h:p" 374 print "test_states.py -ssl [-host=h] [-port=p] -> run the tests using SSL on h:p" 375 sys.exit(0) 376 377 if arg == "-ssl": 378 ssl = True 379 elif arg.startswith("-host="): 380 host = arg[6:].strip("\"'") 381 elif arg.startswith("-port="): 382 port = int(arg[6:].strip()) 383 384 run_all(host, port, ssl) 385