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
43
44 cherrypy.tree.mount(Root())
45 cherrypy.config.update({
46 'environment': 'test_suite',
47 'engine.deadlock_poll_freq': 0.1,
48 })
49
51
53 self.bus = bus
54 self.running = False
55 self.startcount = 0
56 self.gracecount = 0
57 self.threads = {}
58
65
67 self.running = True
68 self.startcount += 1
69
72
75
77 self.threads[thread_id] = None
78
80 del self.threads[thread_id]
81
82 db_connection = Dependency(engine)
83 db_connection.subscribe()
84
85
86
87
88
89
90 from cherrypy.test import helper
91
93
95 if not self.server_class:
96
97
98 self.getPage("/")
99 self.assertStatus(503)
100
101
102 self.assertEqual(db_connection.running, False)
103 self.assertEqual(db_connection.startcount, 0)
104 self.assertEqual(len(db_connection.threads), 0)
105
106
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
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
125 engine.stop()
126 self.assertEqual(engine.state, engine.states.STOPPED)
127
128
129 self.assertEqual(db_connection.running, False)
130 self.assertEqual(len(db_connection.threads), 0)
131
132 if not self.server_class:
133
134
135
136 self.getPage("/")
137 self.assertStatus(503)
138
139
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
150 cherrypy.server.start()
151 engine.start()
152
153
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
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
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
177
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
228
259
289
291 if not self.server_class:
292 print "skipped (no server) ",
293 return
294
295
296
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
306
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
316
317
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
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
330 self.getPage("/exit")
331 p.join()
332
333
334
335 if p.exit_code != 0:
336 self.fail("Daemonized parent process failed to exit cleanly.")
337
338
340
342
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
354 p = helper.CPProcess(ssl=(self.scheme.lower()=='https'))
355 p.write_conf()
356 p.start(imports='cherrypy.test.test_states_demo')
357
358 os.kill(p.get_pid(), SIGHUP)
359
360 p.join()
361
363
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
379
380
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
389 os.kill(pid, SIGHUP)
390
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
398 self.getPage("/exit")
399 p.join()
400
402
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
420 p = helper.CPProcess(ssl=(self.scheme.lower()=='https'))
421 p.write_conf()
422 p.start(imports='cherrypy.test.test_states_demo')
423
424 os.kill(p.get_pid(), SIGTERM)
425
426 p.join()
427
428 if os.name in ['posix']:
429
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
435 os.kill(p.get_pid(), SIGTERM)
436
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
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
461 os.kill(p.get_pid(), SIGTERM)
462
463 p.join()
464
465
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):
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