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
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
64
94
97
100
102 if cherrypy.server.protocol_version == "HTTP/1.1":
103 self.PROTOCOL = "HTTP/1.1"
104
105 self.persistent = True
106
107
108 self.getPage("/")
109 self.assertStatus('200 OK')
110 self.assertBody(pov)
111 self.assertNoHeader("Connection")
112
113
114 if set_cl:
115
116
117 self.getPage("/stream?set_cl=Yes")
118 self.assertHeader("Content-Length")
119 self.assertNoHeader("Connection", "close")
120 self.assertNoHeader("Transfer-Encoding")
121
122 self.assertStatus('200 OK')
123 self.assertBody('0123456789')
124 else:
125
126
127
128 self.getPage("/stream")
129 self.assertNoHeader("Content-Length")
130 self.assertStatus('200 OK')
131 self.assertBody('0123456789')
132
133 chunked_response = False
134 for k, v in self.headers:
135 if k.lower() == "transfer-encoding":
136 if str(v) == "chunked":
137 chunked_response = True
138
139 if chunked_response:
140 self.assertNoHeader("Connection", "close")
141 else:
142 self.assertHeader("Connection", "close")
143
144
145 self.assertRaises(httplib.NotConnected, self.getPage, "/")
146 else:
147 self.PROTOCOL = "HTTP/1.0"
148
149 self.persistent = True
150
151
152 self.getPage("/", headers=[("Connection", "Keep-Alive")])
153 self.assertStatus('200 OK')
154 self.assertBody(pov)
155 self.assertHeader("Connection", "Keep-Alive")
156
157
158 if set_cl:
159
160
161 self.getPage("/stream?set_cl=Yes",
162 headers=[("Connection", "Keep-Alive")])
163 self.assertHeader("Content-Length")
164 self.assertHeader("Connection", "Keep-Alive")
165 self.assertNoHeader("Transfer-Encoding")
166 self.assertStatus('200 OK')
167 self.assertBody('0123456789')
168 else:
169
170
171 self.getPage("/stream", headers=[("Connection", "Keep-Alive")])
172 self.assertStatus('200 OK')
173 self.assertBody('0123456789')
174
175 self.assertNoHeader("Content-Length")
176 self.assertNoHeader("Connection", "Keep-Alive")
177 self.assertNoHeader("Transfer-Encoding")
178
179
180 self.assertRaises(httplib.NotConnected, self.getPage, "/")
181
183 if cherrypy.server.protocol_version != "HTTP/1.1":
184 print "skipped ",
185 return
186
187 old_timeout = None
188 try:
189 httpserver = cherrypy.server.httpservers.keys()[0]
190 old_timeout = httpserver.timeout
191 except (AttributeError, IndexError):
192 print "skipped ",
193 return
194
195 try:
196 httpserver.timeout = timeout
197 self.PROTOCOL = "HTTP/1.1"
198
199
200 self.persistent = True
201 conn = self.HTTP_CONN
202 conn.putrequest("GET", "/", skip_host=True)
203 conn.putheader("Host", self.HOST)
204 conn.endheaders()
205 response = conn.response_class(conn.sock, method="GET")
206 response.begin()
207 self.assertEqual(response.status, 200)
208 self.body = response.read()
209 self.assertBody(pov)
210
211
212 conn._output('GET /hello HTTP/1.1')
213 conn._output("Host: %s" % self.HOST)
214 conn._send_output()
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("Hello, world!")
220
221
222 time.sleep(timeout * 2)
223
224
225 conn._output('GET /hello HTTP/1.1')
226 conn._output("Host: %s" % self.HOST)
227 conn._send_output()
228 response = conn.response_class(conn.sock, method="GET")
229 try:
230 response.begin()
231 except:
232 if not isinstance(sys.exc_info()[1],
233 (socket.error, httplib.BadStatusLine)):
234 self.fail("Writing to timed out socket didn't fail"
235 " as it should have: %s" % sys.exc_info()[1])
236 else:
237 self.fail("Writing to timed out socket didn't fail"
238 " as it should have: %s" %
239 response.read())
240
241 conn.close()
242
243
244 self.persistent = True
245 conn = self.HTTP_CONN
246 conn.putrequest("GET", "/", skip_host=True)
247 conn.putheader("Host", self.HOST)
248 conn.endheaders()
249 response = conn.response_class(conn.sock, method="GET")
250 response.begin()
251 self.assertEqual(response.status, 200)
252 self.body = response.read()
253 self.assertBody(pov)
254 finally:
255 if old_timeout is not None:
256 httpserver.timeout = old_timeout
257
259 if cherrypy.server.protocol_version != "HTTP/1.1":
260 print "skipped ",
261 return
262
263 self.PROTOCOL = "HTTP/1.1"
264
265
266 self.persistent = True
267 conn = self.HTTP_CONN
268
269
270 conn.putrequest("GET", "/hello", skip_host=True)
271 conn.putheader("Host", self.HOST)
272 conn.endheaders()
273
274 for trial in xrange(5):
275
276 conn._output('GET /hello HTTP/1.1')
277 conn._output("Host: %s" % self.HOST)
278 conn._send_output()
279
280
281 response = conn.response_class(conn.sock, method="GET")
282 response.begin()
283 body = response.read()
284 self.assertEqual(response.status, 200)
285 self.assertEqual(body, "Hello, world!")
286
287
288 response = conn.response_class(conn.sock, method="GET")
289 response.begin()
290 body = response.read()
291 self.assertEqual(response.status, 200)
292 self.assertEqual(body, "Hello, world!")
293
294 conn.close()
295
297 if cherrypy.server.protocol_version != "HTTP/1.1":
298 print "skipped ",
299 return
300
301 self.PROTOCOL = "HTTP/1.1"
302
303 self.persistent = True
304 conn = self.HTTP_CONN
305
306
307
308
309 conn.putrequest("POST", "/upload", skip_host=True)
310 conn.putheader("Host", self.HOST)
311 conn.putheader("Content-Type", "text/plain")
312 conn.putheader("Content-Length", "4")
313 conn.endheaders()
314 conn.send("d'oh")
315 response = conn.response_class(conn.sock, method="POST")
316 version, status, reason = response._read_status()
317 self.assertNotEqual(status, 100)
318 conn.close()
319
320
321 conn.connect()
322 conn.putrequest("POST", "/upload", skip_host=True)
323 conn.putheader("Host", self.HOST)
324 conn.putheader("Content-Type", "text/plain")
325 conn.putheader("Content-Length", "17")
326 conn.putheader("Expect", "100-continue")
327 conn.endheaders()
328 response = conn.response_class(conn.sock, method="POST")
329
330
331 version, status, reason = response._read_status()
332 self.assertEqual(status, 100)
333 while True:
334 skip = response.fp.readline().strip()
335 if not skip:
336 break
337
338
339 conn.send("I am a small file")
340
341
342 response.begin()
343 self.status, self.headers, self.body = webtest.shb(response)
344 self.assertStatus(200)
345 self.assertBody("thanks for 'I am a small file' (text/plain)")
346
348 if cherrypy.server.protocol_version != "HTTP/1.1":
349 print "skipped ",
350 return
351
352 self.PROTOCOL = "HTTP/1.1"
353
354
355 self.persistent = True
356 conn = self.HTTP_CONN
357
358
359 self.getPage("/")
360 self.assertStatus('200 OK')
361 self.assertBody(pov)
362 self.assertNoHeader("Connection")
363
364
365 self.getPage("/custom/204")
366 self.assertStatus(204)
367 self.assertNoHeader("Content-Length")
368 self.assertBody("")
369 self.assertNoHeader("Connection")
370
371
372 self.getPage("/custom/304")
373 self.assertStatus(304)
374 self.assertNoHeader("Content-Length")
375 self.assertBody("")
376 self.assertNoHeader("Connection")
377
379 if cherrypy.server.protocol_version != "HTTP/1.1":
380 print "skipped ",
381 return
382
383 if (hasattr(self, 'harness') and
384 "modpython" in self.harness.__class__.__name__.lower()):
385
386 print "skipped ",
387 return
388
389 self.PROTOCOL = "HTTP/1.1"
390
391
392 self.persistent = True
393 conn = self.HTTP_CONN
394
395
396 body = ("8;key=value\r\nxx\r\nxxxx\r\n5\r\nyyyyy\r\n0\r\n"
397 "Content-Type: application/x-json\r\n\r\n")
398 conn.putrequest("POST", "/upload", skip_host=True)
399 conn.putheader("Host", self.HOST)
400 conn.putheader("Transfer-Encoding", "chunked")
401 conn.putheader("Trailer", "Content-Type")
402
403
404
405 conn.putheader("Content-Length", len(body))
406 conn.endheaders()
407 conn.send(body)
408 response = conn.getresponse()
409 self.status, self.headers, self.body = webtest.shb(response)
410 self.assertStatus('200 OK')
411 self.assertBody("thanks for 'xx\r\nxxxxyyyyy' (application/x-json)")
412
413
414
415 body = "5f\r\n" + ("x" * 95) + "\r\n0\r\n\r\n"
416 conn.putrequest("POST", "/upload", skip_host=True)
417 conn.putheader("Host", self.HOST)
418 conn.putheader("Transfer-Encoding", "chunked")
419 conn.putheader("Content-Type", "text/plain")
420
421 conn.endheaders()
422 conn.send(body)
423 response = conn.getresponse()
424 self.status, self.headers, self.body = webtest.shb(response)
425 self.assertStatus(413)
426 self.assertBody("")
427
454
455
456
457
458 if __name__ == "__main__":
459 setup_server()
460 helper.testmain()
461