Package cherrypy :: Package test :: Module test_conn
[hide private]
[frames] | no frames]

Source Code for Module cherrypy.test.test_conn

  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 self.PROTOCOL = "HTTP/1.1" 104 105 self.persistent = True 106 107 # Make the first request and assert there's no "Connection: close". 108 self.getPage("/") 109 self.assertStatus('200 OK') 110 self.assertBody(pov) 111 self.assertNoHeader("Connection") 112 113 # Make another, streamed request on the same connection. 114 if set_cl: 115 # When a Content-Length is provided, the content should stream 116 # without closing the connection. 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 # 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 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 # Make another request on the same connection, which should error. 145 self.assertRaises(httplib.NotConnected, self.getPage, "/") 146 else: 147 self.PROTOCOL = "HTTP/1.0" 148 149 self.persistent = True 150 151 # Make the first request and assert Keep-Alive. 152 self.getPage("/", headers=[("Connection", "Keep-Alive")]) 153 self.assertStatus('200 OK') 154 self.assertBody(pov) 155 self.assertHeader("Connection", "Keep-Alive") 156 157 # Make another, streamed request on the same connection. 158 if set_cl: 159 # When a Content-Length is provided, the content should 160 # stream without closing the connection. 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 # When a Content-Length is not provided, 170 # the server should close the connection. 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 # Make another request on the same connection, which should error. 180 self.assertRaises(httplib.NotConnected, self.getPage, "/")
181
182 - def test_HTTP11_Timeout(self):
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 # Make an initial request 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 # Make a second request on the same socket 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 # Wait for our socket timeout 222 time.sleep(timeout * 2) 223 224 # Make another request on the same socket, which should error 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 # Make another request on a new socket, which should work 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
258 - def test_HTTP11_pipelining(self):
259 if cherrypy.server.protocol_version != "HTTP/1.1": 260 print "skipped ", 261 return 262 263 self.PROTOCOL = "HTTP/1.1" 264 265 # Test pipelining. httplib doesn't support this directly. 266 self.persistent = True 267 conn = self.HTTP_CONN 268 269 # Put request 1 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 # Put next request 276 conn._output('GET /hello HTTP/1.1') 277 conn._output("Host: %s" % self.HOST) 278 conn._send_output() 279 280 # Retrieve previous response 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 # Retrieve final response 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
296 - def test_100_Continue(self):
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 # Try a page without an Expect request header first. 307 # Note that httplib's response.begin automatically ignores 308 # 100 Continue responses, so we must manually check for it. 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 # Now try a page with an Expect header... 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 # ...assert and then skip the 100 response 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 # ...send the body 339 conn.send("I am a small file") 340 341 # ...get the final response 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
347 - def test_No_Message_Body(self):
348 if cherrypy.server.protocol_version != "HTTP/1.1": 349 print "skipped ", 350 return 351 352 self.PROTOCOL = "HTTP/1.1" 353 354 # Set our HTTP_CONN to an instance so it persists between requests. 355 self.persistent = True 356 conn = self.HTTP_CONN 357 358 # Make the first request and assert there's no "Connection: close". 359 self.getPage("/") 360 self.assertStatus('200 OK') 361 self.assertBody(pov) 362 self.assertNoHeader("Connection") 363 364 # Make a 204 request on the same connection. 365 self.getPage("/custom/204") 366 self.assertStatus(204) 367 self.assertNoHeader("Content-Length") 368 self.assertBody("") 369 self.assertNoHeader("Connection") 370 371 # Make a 304 request on the same connection. 372 self.getPage("/custom/304") 373 self.assertStatus(304) 374 self.assertNoHeader("Content-Length") 375 self.assertBody("") 376 self.assertNoHeader("Connection")
377
378 - def test_Chunked_Encoding(self):
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 # mod_python forbids chunked encoding 386 print "skipped ", 387 return 388 389 self.PROTOCOL = "HTTP/1.1" 390 391 # Set our HTTP_CONN to an instance so it persists between requests. 392 self.persistent = True 393 conn = self.HTTP_CONN 394 395 # Try a normal chunked request (with extensions) 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 # Note that this is somewhat malformed: 403 # we shouldn't be sending Content-Length. 404 # RFC 2616 says the server should ignore it. 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 # Try a chunked request that exceeds server.max_request_body_size. 414 # Note that the delimiters and trailer are included. 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 ## conn.putheader("Content-Length", len(body)) 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
428 - def test_HTTP10(self):
429 self.PROTOCOL = "HTTP/1.0" 430 if self.scheme == "https": 431 self.HTTP_CONN = httplib.HTTPSConnection 432 else: 433 self.HTTP_CONN = httplib.HTTPConnection 434 435 # Test a normal HTTP/1.0 request. 436 self.getPage("/page2") 437 self.assertStatus('200 OK') 438 self.assertBody(pov) 439 # Apache, for example, may emit a Connection header even for HTTP/1.0 440 ## self.assertNoHeader("Connection") 441 442 # Test a keep-alive HTTP/1.0 request. 443 self.persistent = True 444 445 self.getPage("/page3", headers=[("Connection", "Keep-Alive")]) 446 self.assertStatus('200 OK') 447 self.assertBody(pov) 448 self.assertHeader("Connection", "Keep-Alive") 449 450 # Remove the keep-alive header again. 451 self.getPage("/page3") 452 self.assertStatus('200 OK') 453 self.assertBody(pov)
454 # Apache, for example, may emit a Connection header even for HTTP/1.0 455 ## self.assertNoHeader("Connection") 456 457 458 if __name__ == "__main__": 459 setup_server() 460 helper.testmain() 461