Revision Log

test_states.py

Revision 1475 (checked in by fumanchu, 2 years ago)	Various tweaks to test_states.
Property svn:eol-style set to `native`

Line
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)

Note: See TracBrowser for help on using the browser.

root/tags/cherrypy-3.0.0/cherrypy/test/test_states.py

Download in other formats: