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
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
50
52 self.running = False
53 self.startcount = 0
54 self.threads = {}
55
57 self.running = True
58 self.startcount += 1
59
62
64 self.threads[thread_id] = None
65
67 del self.threads[thread_id]
68
69
70 from cherrypy.test import helper
71
73
75 if not self.server_class:
76
77
78 self.getPage("/")
79 self.assertStatus(503)
80
81
82 self.assertEqual(db_connection.running, False)
83 self.assertEqual(db_connection.startcount, 0)
84 self.assertEqual(len(db_connection.threads), 0)
85
86
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
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
106 cherrypy.engine.stop()
107 self.assertEqual(cherrypy.engine.state, 0)
108
109
110 self.assertEqual(db_connection.running, False)
111 self.assertEqual(len(db_connection.threads), 0)
112
113 if not self.server_class:
114
115
116
117 self.getPage("/")
118 self.assertStatus(503)
119
120
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
130 cherrypy.server.start()
131 cherrypy.engine.start(blocking=False)
132
133
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
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
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
157
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
211
240
242 if not self.server_class:
243 print "skipped (no server) ",
244 return
245
246
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
266 time.sleep(2)
267
268
269 f = open(demoscript, 'ab')
270 f.write(" ")
271 f.close()
272
273
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
283 self.getPage("/stop")
284
285 try:
286 try:
287
288 print os.wait()
289 except AttributeError:
290
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):
326
327
328 -def run_all(host, port, ssl=False):
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