Revision Log

test_conn.py

Revision 1550 (checked in by fumanchu, 2 years ago)	Better solution to #625: have the SSL file objects inheirt the timeout of their socket, which is copied from httpserver.timeout.
Property svn:eol-style set to `native`

Line
1	from cherrypy.test import test
2	test.prefer_parent_path()
3
4	import httplib
5	import socket
6	import sys
7	import time
8	timeout = 0.1
9
10
11	import cherrypy
12	from cherrypy.test import webtest
13
14
15	pov = 'pPeErRsSiIsStTeEnNcCeE oOfF vViIsSiIoOnN'
16
17	def setup_server():
18	class Root:
19
20	def index(self):
21	return pov
22	index.exposed = True
23	page1 = index
24	page2 = index
25	page3 = index
26
27	def hello(self):
28	return "Hello, world!"
29	hello.exposed = True
30
31	def stream(self, set_cl=False):
32	if set_cl:
33	cherrypy.response.headers['Content-Length'] = 10
34
35	def content():
36	for x in xrange(10):
37	yield str(x)
38
39	return content()
40	stream.exposed = True
41	stream._cp_config = {'response.stream': True}
42
43	def upload(self):
44	return ("thanks for '%s' (%s)" %
45	(cherrypy.request.body.read(),
46	cherrypy.request.headers['Content-Type']))
47	upload.exposed = True
48
49	def custom(self, response_code):
50	cherrypy.response.status = response_code
51	return "Code = %s" % response_code
52	custom.exposed = True
53
54	cherrypy.tree.mount(Root())
55	cherrypy.config.update({
56	'server.max_request_body_size': 100,
57	'environment': 'test_suite',
58	})
59
60
61	from cherrypy.test import helper
62
63	class ConnectionTests(helper.CPWebCase):
64
65	def test_HTTP11(self):
66	if cherrypy.server.protocol_version != "HTTP/1.1":
67	print "skipped ",
68	return
69
70	self.PROTOCOL = "HTTP/1.1"
71
72	self.persistent = True
73
74	# Make the first request and assert there's no "Connection: close".
75	self.getPage("/")
76	self.assertStatus('200 OK')
77	self.assertBody(pov)
78	self.assertNoHeader("Connection")
79
80	# Make another request on the same connection.
81	self.getPage("/page1")
82	self.assertStatus('200 OK')
83	self.assertBody(pov)
84	self.assertNoHeader("Connection")
85
86	# Test client-side close.
87	self.getPage("/page2", headers=[("Connection", "close")])
88	self.assertStatus('200 OK')
89	self.assertBody(pov)
90	self.assertHeader("Connection", "close")
91
92	# Make another request on the same connection, which should error.
93	self.assertRaises(httplib.NotConnected, self.getPage, "/")
94
95	def test_Streaming_no_len(self):
96	self._streaming(set_cl=False)
97
98	def test_Streaming_with_len(self):
99	self._streaming(set_cl=True)
100
101	def _streaming(self, set_cl):
102	if cherrypy.server.protocol_version != "HTTP/1.1":
103	print "skipped ",
104	return
105
106	self.PROTOCOL = "HTTP/1.1"
107
108	self.persistent = True
109
110	# Make the first request and assert there's no "Connection: close".
111	self.getPage("/")
112	self.assertStatus('200 OK')
113	self.assertBody(pov)
114	self.assertNoHeader("Connection")
115
116	# Make another, streamed request on the same connection.
117	if set_cl:
118	# When a Content-Length is provided, the content should stream
119	# without closing the connection.
120	self.getPage("/stream?set_cl=Yes")
121	self.assertHeader("Content-Length")
122	self.assertNoHeader("Connection", "close")
123	self.assertNoHeader("Transfer-Encoding")
124	else:
125	# When no Content-Length response header is provided,
126	# streamed output will either close the connection, or use
127	# chunked encoding, to determine transfer-length.
128	self.getPage("/stream")
129	self.assertNoHeader("Content-Length")
130
131	chunked_response = False
132	for k, v in self.headers:
133	if k.lower() == "transfer-encoding":
134	if str(v) == "chunked":
135	chunked_response = True
136
137	if chunked_response:
138	self.assertNoHeader("Connection", "close")
139	else:
140	self.assertHeader("Connection", "close")
141
142	# Make another request on the same connection, which should error.
143	self.assertRaises(httplib.NotConnected, self.getPage, "/")
144
145	self.assertStatus('200 OK')
146	self.assertBody('0123456789')
147
148	def test_HTTP11_Timeout(self):
149	if cherrypy.server.protocol_version != "HTTP/1.1":
150	print "skipped ",
151	return
152
153	old_timeout = None
154	try:
155	httpserver = cherrypy.server.httpservers.keys()[0]
156	old_timeout = httpserver.timeout
157	except (AttributeError, IndexError):
158	print "skipped ",
159	return
160
161	try:
162	httpserver.timeout = timeout
163	self.PROTOCOL = "HTTP/1.1"
164
165	# Make an initial request
166	self.persistent = True
167	conn = self.HTTP_CONN
168	conn.putrequest("GET", "/", skip_host=True)
169	conn.putheader("Host", self.HOST)
170	conn.endheaders()
171	response = conn.response_class(conn.sock, method="GET")
172	response.begin()
173	self.assertEqual(response.status, 200)
174	self.body = response.read()
175	self.assertBody(pov)
176
177	# Make a second request on the same socket
178	conn._output('GET /hello HTTP/1.1')
179	conn._output("Host: %s" % self.HOST)
180	conn._send_output()
181	response = conn.response_class(conn.sock, method="GET")
182	response.begin()
183	self.assertEqual(response.status, 200)
184	self.body = response.read()
185	self.assertBody("Hello, world!")
186
187	# Wait for our socket timeout
188	time.sleep(timeout * 2)
189
190	# Make another request on the same socket, which should error
191	conn._output('GET /hello HTTP/1.1')
192	conn._output("Host: %s" % self.HOST)
193	conn._send_output()
194	response = conn.response_class(conn.sock, method="GET")
195	try:
196	response.begin()
197	except:
198	if not isinstance(sys.exc_info()[1],
199	(socket.error, httplib.BadStatusLine)):
200	self.fail("Writing to timed out socket didn't fail"
201	" as it should have: %s" % sys.exc_info()[1])
202	else:
203	self.fail("Writing to timed out socket didn't fail"
204	" as it should have: %s" %
205	response.read())
206
207	conn.close()
208
209	# Make another request on a new socket, which should work
210	self.persistent = True
211	conn = self.HTTP_CONN
212	conn.putrequest("GET", "/", skip_host=True)
213	conn.putheader("Host", self.HOST)
214	conn.endheaders()
215	response = conn.response_class(conn.sock, method="GET")
216	response.begin()
217	self.assertEqual(response.status, 200)
218	self.body = response.read()
219	self.assertBody(pov)
220	finally:
221	if old_timeout is not None:
222	httpserver.timeout = old_timeout
223
224	def test_HTTP11_pipelining(self):
225	if cherrypy.server.protocol_version != "HTTP/1.1":
226	print "skipped ",
227	return
228
229	self.PROTOCOL = "HTTP/1.1"
230
231	# Test pipelining. httplib doesn't support this directly.
232	self.persistent = True
233	conn = self.HTTP_CONN
234
235	# Put request 1
236	conn.putrequest("GET", "/hello", skip_host=True)
237	conn.putheader("Host", self.HOST)
238	conn.endheaders()
239
240	for trial in xrange(5):
241	# Put next request
242	conn._output('GET /hello HTTP/1.1')
243	conn._output("Host: %s" % self.HOST)
244	conn._send_output()
245
246	# Retrieve previous response
247	response = conn.response_class(conn.sock, method="GET")
248	response.begin()
249	body = response.read()
250	self.assertEqual(response.status, 200)
251	self.assertEqual(body, "Hello, world!")
252
253	# Retrieve final response
254	response = conn.response_class(conn.sock, method="GET")
255	response.begin()
256	body = response.read()
257	self.assertEqual(response.status, 200)
258	self.assertEqual(body, "Hello, world!")
259
260	conn.close()
261
262	def test_100_Continue(self):
263	if cherrypy.server.protocol_version != "HTTP/1.1":
264	print "skipped ",
265	return
266
267	self.PROTOCOL = "HTTP/1.1"
268
269	self.persistent = True
270	conn = self.HTTP_CONN
271
272	# Try a page without an Expect request header first.
273	# Note that httplib's response.begin automatically ignores
274	# 100 Continue responses, so we must manually check for it.
275	conn.putrequest("POST", "/upload", skip_host=True)
276	conn.putheader("Host", self.HOST)
277	conn.putheader("Content-Type", "text/plain")
278	conn.putheader("Content-Length", "4")
279	conn.endheaders()
280	conn.send("d'oh")
281	response = conn.response_class(conn.sock, method="POST")
282	version, status, reason = response._read_status()
283	self.assertNotEqual(status, 100)
284	conn.close()
285
286	# Now try a page with an Expect header...
287	conn.connect()
288	conn.putrequest("POST", "/upload", skip_host=True)
289	conn.putheader("Host", self.HOST)
290	conn.putheader("Content-Type", "text/plain")
291	conn.putheader("Content-Length", "17")
292	conn.putheader("Expect", "100-continue")
293	conn.endheaders()
294	response = conn.response_class(conn.sock, method="POST")
295
296	# ...assert and then skip the 100 response
297	version, status, reason = response._read_status()
298	self.assertEqual(status, 100)
299	while True:
300	skip = response.fp.readline().strip()
301	if not skip:
302	break
303
304	# ...send the body
305	conn.send("I am a small file")
306
307	# ...get the final response
308	response.begin()
309	self.status, self.headers, self.body = webtest.shb(response)
310	self.assertStatus(200)
311	self.assertBody("thanks for 'I am a small file' (text/plain)")
312
313	def test_No_Message_Body(self):
314	if cherrypy.server.protocol_version != "HTTP/1.1":
315	print "skipped ",
316	return
317
318	self.PROTOCOL = "HTTP/1.1"
319
320	# Set our HTTP_CONN to an instance so it persists between requests.
321	self.persistent = True
322	conn = self.HTTP_CONN
323
324	# Make the first request and assert there's no "Connection: close".
325	self.getPage("/")
326	self.assertStatus('200 OK')
327	self.assertBody(pov)
328	self.assertNoHeader("Connection")
329
330	# Make a 204 request on the same connection.
331	self.getPage("/custom/204")
332	self.assertStatus(204)
333	self.assertNoHeader("Content-Length")
334	self.assertBody("")
335	self.assertNoHeader("Connection")
336
337	# Make a 304 request on the same connection.
338	self.getPage("/custom/304")
339	self.assertStatus(304)
340	self.assertNoHeader("Content-Length")
341	self.assertBody("")
342	self.assertNoHeader("Connection")
343
344	def test_Chunked_Encoding(self):
345	if cherrypy.server.protocol_version != "HTTP/1.1":
346	print "skipped ",
347	return
348
349	if (hasattr(self, 'harness') and
350	"modpython" in self.harness.__class__.__name__.lower()):
351	# mod_python forbids chunked encoding
352	print "skipped ",
353	return
354
355	self.PROTOCOL = "HTTP/1.1"
356
357	# Set our HTTP_CONN to an instance so it persists between requests.
358	self.persistent = True
359	conn = self.HTTP_CONN
360
361	# Try a normal chunked request (with extensions)
362	body = ("8;key=value\r\nxx\r\nxxxx\r\n5\r\nyyyyy\r\n0\r\n"
363	"Content-Type: application/x-json\r\n\r\n")
364	conn.putrequest("POST", "/upload", skip_host=True)
365	conn.putheader("Host", self.HOST)
366	conn.putheader("Transfer-Encoding", "chunked")
367	conn.putheader("Trailer", "Content-Type")
368	# Note that this is somewhat malformed:
369	# we shouldn't be sending Content-Length.
370	# RFC 2616 says the server should ignore it.
371	conn.putheader("Content-Length", len(body))
372	conn.endheaders()
373	conn.send(body)
374	response = conn.getresponse()
375	self.status, self.headers, self.body = webtest.shb(response)
376	self.assertStatus('200 OK')
377	self.assertBody("thanks for 'xx\r\nxxxxyyyyy' (application/x-json)")
378
379	# Try a chunked request that exceeds server.max_request_body_size.
380	# Note that the delimiters and trailer are included.
381	body = "5f\r\n" + ("x" * 95) + "\r\n0\r\n\r\n"
382	conn.putrequest("POST", "/upload", skip_host=True)
383	conn.putheader("Host", self.HOST)
384	conn.putheader("Transfer-Encoding", "chunked")
385	conn.putheader("Content-Type", "text/plain")
386	## conn.putheader("Content-Length", len(body))
387	conn.endheaders()
388	conn.send(body)
389	response = conn.getresponse()
390	self.status, self.headers, self.body = webtest.shb(response)
391	self.assertStatus(413)
392	self.assertBody("")
393
394	def test_HTTP10(self):
395	self.PROTOCOL = "HTTP/1.0"
396	if self.scheme == "https":
397	self.HTTP_CONN = httplib.HTTPSConnection
398	else:
399	self.HTTP_CONN = httplib.HTTPConnection
400
401	# Test a normal HTTP/1.0 request.
402	self.getPage("/page2")
403	self.assertStatus('200 OK')
404	self.assertBody(pov)
405	# Apache, for example, may emit a Connection header even for HTTP/1.0
406	## self.assertNoHeader("Connection")
407
408	# Test a keep-alive HTTP/1.0 request.
409	self.persistent = True
410
411	self.getPage("/page3", headers=[("Connection", "Keep-Alive")])
412	self.assertStatus('200 OK')
413	self.assertBody(pov)
414	self.assertHeader("Connection", "Keep-Alive")
415
416	# Remove the keep-alive header again.
417	self.getPage("/page3")
418	self.assertStatus('200 OK')
419	self.assertBody(pov)
420	# Apache, for example, may emit a Connection header even for HTTP/1.0
421	## self.assertNoHeader("Connection")
422
423
424	if __name__ == "__main__":
425	setup_server()
426	helper.testmain()

Note: See TracBrowser for help on using the browser.

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

Download in other formats: