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
13
15 self.bus = bus
16 self.running = False
17 self.startcount = 0
18 self.gracecount = 0
19 self.threads = {}
20
27
29 self.running = True
30 self.startcount += 1
31
34
37
40
43
44 db_connection = Dependency(engine)
45
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
85
86
87 from cherrypy.test import helper
88
90 setup_server = staticmethod(setup_server)
91
95
97 engine.stop()
98
99 self.assertEqual(db_connection.running, False)
100 self.assertEqual(db_connection.startcount, 1)
101 self.assertEqual(len(db_connection.threads), 0)
102
103
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
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
121 engine.stop()
122 self.assertEqual(engine.state, engine.states.STOPPED)
123
124
125 self.assertEqual(db_connection.running, False)
126 self.assertEqual(len(db_connection.threads), 0)
127
128
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
139 cherrypy.server.start()
140 engine.start()
141
142
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
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
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
166
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
215
246
277
279
280
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
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
300
301
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
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
317 self.getPage("/exit")
318 p.join()
319
320
321
322 if p.exit_code != 0:
323 self.fail("Daemonized parent process failed to exit cleanly.")
324
325
328
329 try:
330 from signal import SIGHUP
331 except ImportError:
332 return self.skip("skipped (no SIGHUP) ")
333
334
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
340 os.kill(p.get_pid(), SIGHUP)
341
342 p.join()
343
345
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
355
356
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
366 os.kill(pid, SIGHUP)
367
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
375 self.getPage("/exit")
376 p.join()
377
379
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
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
396 os.kill(p.get_pid(), SIGTERM)
397
398 p.join()
399
400 if os.name in ['posix']:
401
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
408 os.kill(p.get_pid(), SIGTERM)
409
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
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
431 os.kill(p.get_pid(), SIGTERM)
432
433 p.join()
434
435
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