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

Source Code for Module cherrypy.test.test_conn

  1  """Tests for TCP connection handling, including proper and timely close.""" 
  2   
  3  import socket 
  4  import sys 
  5  import time 
  6  timeout = 1 
  7   
  8   
  9  import cherrypy 
 10  from cherrypy._cpcompat import HTTPConnection, HTTPSConnection, NotConnected 
 11  from cherrypy._cpcompat import BadStatusLine, ntob, urlopen, unicodestr 
 12  from cherrypy.test import webtest 
 13  from cherrypy import _cperror 
 14   
 15   
 16  pov = 'pPeErRsSiIsStTeEnNcCeE oOfF vViIsSiIoOnN' 
 17   
 18   
19 -def setup_server():
20 21 def raise500(): 22 raise cherrypy.HTTPError(500)
23 24 class Root: 25 26 def index(self): 27 return pov 28 index.exposed = True 29 page1 = index 30 page2 = index 31 page3 = index 32 33 def hello(self): 34 return "Hello, world!" 35 hello.exposed = True 36 37 def timeout(self, t): 38 return str(cherrypy.server.httpserver.timeout) 39 timeout.exposed = True 40 41 def stream(self, set_cl=False): 42 if set_cl: 43 cherrypy.response.headers['Content-Length'] = 10 44 45 def content(): 46 for x in range(10): 47 yield str(x) 48 49 return content() 50 stream.exposed = True 51 stream._cp_config = {'response.stream': True} 52 53 def error(self, code=500): 54 raise cherrypy.HTTPError(code) 55 error.exposed = True 56 57 def upload(self): 58 if not cherrypy.request.method == 'POST': 59 raise AssertionError("'POST' != request.method %r" % 60 cherrypy.request.method) 61 return "thanks for '%s'" % cherrypy.request.body.read() 62 upload.exposed = True 63 64 def custom(self, response_code): 65 cherrypy.response.status = response_code 66 return "Code = %s" % response_code 67 custom.exposed = True 68 69 def err_before_read(self): 70 return "ok" 71 err_before_read.exposed = True 72 err_before_read._cp_config = {'hooks.on_start_resource': raise500} 73 74 def one_megabyte_of_a(self): 75 return ["a" * 1024] * 1024 76 one_megabyte_of_a.exposed = True 77 78 def custom_cl(self, body, cl): 79 cherrypy.response.headers['Content-Length'] = cl 80 if not isinstance(body, list): 81 body = [body] 82 newbody = [] 83 for chunk in body: 84 if isinstance(chunk, unicodestr): 85 chunk = chunk.encode('ISO-8859-1') 86 newbody.append(chunk) 87 return newbody 88 custom_cl.exposed = True 89 # Turn off the encoding tool so it doens't collapse 90 # our response body and reclaculate the Content-Length. 91 custom_cl._cp_config = {'tools.encode.on': False} 92 93 cherrypy.tree.mount(Root()) 94 cherrypy.config.update({ 95 'server.max_request_body_size': 1001, 96 'server.socket_timeout': timeout, 97 }) 98 99 100 from cherrypy.test import helper 101 102
103 -class ConnectionCloseTests(helper.CPWebCase):
104 setup_server = staticmethod(setup_server) 105
106 - def test_HTTP11(self):
107 if cherrypy.server.protocol_version != "HTTP/1.1": 108 return self.skip() 109 110 self.PROTOCOL = "HTTP/1.1" 111 112 self.persistent = True 113 114 # Make the first request and assert there's no "Connection: close". 115 self.getPage("/") 116 self.assertStatus('200 OK') 117 self.assertBody(pov) 118 self.assertNoHeader("Connection") 119 120 # Make another request on the same connection. 121 self.getPage("/page1") 122 self.assertStatus('200 OK') 123 self.assertBody(pov) 124 self.assertNoHeader("Connection") 125 126 # Test client-side close. 127 self.getPage("/page2", headers=[("Connection", "close")]) 128 self.assertStatus('200 OK') 129 self.assertBody(pov) 130 self.assertHeader("Connection", "close") 131 132 # Make another request on the same connection, which should error. 133 self.assertRaises(NotConnected, self.getPage, "/")
134
135 - def test_Streaming_no_len(self):
136 self._streaming(set_cl=False)
137
138 - def test_Streaming_with_len(self):
139 self._streaming(set_cl=True)
140
141 - def _streaming(self, set_cl):
142 if cherrypy.server.protocol_version == "HTTP/1.1": 143 self.PROTOCOL = "HTTP/1.1" 144 145 self.persistent = True 146 147 # Make the first request and assert there's no "Connection: close". 148 self.getPage("/") 149 self.assertStatus('200 OK') 150 self.assertBody(pov) 151 self.assertNoHeader("Connection") 152 153 # Make another, streamed request on the same connection. 154 if set_cl: 155 # When a Content-Length is provided, the content should stream 156 # without closing the connection. 157 self.getPage("/stream?set_cl=Yes") 158 self.assertHeader("Content-Length") 159 self.assertNoHeader("Connection", "close") 160 self.assertNoHeader("Transfer-Encoding") 161 162 self.assertStatus('200 OK') 163 self.assertBody('0123456789') 164 else: 165 # When no Content-Length response header is provided, 166 # streamed output will either close the connection, or use 167 # chunked encoding, to determine transfer-length. 168 self.getPage("/stream") 169 self.assertNoHeader("Content-Length") 170 self.assertStatus('200 OK') 171 self.assertBody('0123456789') 172 173 chunked_response = False 174 for k, v in self.headers: 175 if k.lower() == "transfer-encoding": 176 if str(v) == "chunked": 177 chunked_response = True 178 179 if chunked_response: 180 self.assertNoHeader("Connection", "close") 181 else: 182 self.assertHeader("Connection", "close") 183 184 # Make another request on the same connection, which should 185 # error. 186 self.assertRaises(NotConnected, self.getPage, "/") 187 188 # Try HEAD. See 189 # https://bitbucket.org/cherrypy/cherrypy/issue/864. 190 self.getPage("/stream", method='HEAD') 191 self.assertStatus('200 OK') 192 self.assertBody('') 193 self.assertNoHeader("Transfer-Encoding") 194 else: 195 self.PROTOCOL = "HTTP/1.0" 196 197 self.persistent = True 198 199 # Make the first request and assert Keep-Alive. 200 self.getPage("/", headers=[("Connection", "Keep-Alive")]) 201 self.assertStatus('200 OK') 202 self.assertBody(pov) 203 self.assertHeader("Connection", "Keep-Alive") 204 205 # Make another, streamed request on the same connection. 206 if set_cl: 207 # When a Content-Length is provided, the content should 208 # stream without closing the connection. 209 self.getPage("/stream?set_cl=Yes", 210 headers=[("Connection", "Keep-Alive")]) 211 self.assertHeader("Content-Length") 212 self.assertHeader("Connection", "Keep-Alive") 213 self.assertNoHeader("Transfer-Encoding") 214 self.assertStatus('200 OK') 215 self.assertBody('0123456789') 216 else: 217 # When a Content-Length is not provided, 218 # the server should close the connection. 219 self.getPage("/stream", headers=[("Connection", "Keep-Alive")]) 220 self.assertStatus('200 OK') 221 self.assertBody('0123456789') 222 223 self.assertNoHeader("Content-Length") 224 self.assertNoHeader("Connection", "Keep-Alive") 225 self.assertNoHeader("Transfer-Encoding") 226 227 # Make another request on the same connection, which should 228 # error. 229 self.assertRaises(NotConnected, self.getPage, "/")
230
231 - def test_HTTP10_KeepAlive(self):
232 self.PROTOCOL = "HTTP/1.0" 233 if self.scheme == "https": 234 self.HTTP_CONN = HTTPSConnection 235 else: 236 self.HTTP_CONN = HTTPConnection 237 238 # Test a normal HTTP/1.0 request. 239 self.getPage("/page2") 240 self.assertStatus('200 OK') 241 self.assertBody(pov) 242 # Apache, for example, may emit a Connection header even for HTTP/1.0 243 # self.assertNoHeader("Connection") 244 245 # Test a keep-alive HTTP/1.0 request. 246 self.persistent = True 247 248 self.getPage("/page3", headers=[("Connection", "Keep-Alive")]) 249 self.assertStatus('200 OK') 250 self.assertBody(pov) 251 self.assertHeader("Connection", "Keep-Alive") 252 253 # Remove the keep-alive header again. 254 self.getPage("/page3") 255 self.assertStatus('200 OK') 256 self.assertBody(pov)
257 # Apache, for example, may emit a Connection header even for HTTP/1.0 258 # self.assertNoHeader("Connection") 259 260
261 -class PipelineTests(helper.CPWebCase):
262 setup_server = staticmethod(setup_server) 263
264 - def test_HTTP11_Timeout(self):
265 # If we timeout without sending any data, 266 # the server will close the conn with a 408. 267 if cherrypy.server.protocol_version != "HTTP/1.1": 268 return self.skip() 269 270 self.PROTOCOL = "HTTP/1.1" 271 272 # Connect but send nothing. 273 self.persistent = True 274 conn = self.HTTP_CONN 275 conn.auto_open = False 276 conn.connect() 277 278 # Wait for our socket timeout 279 time.sleep(timeout * 2) 280 281 # The request should have returned 408 already. 282 response = conn.response_class(conn.sock, method="GET") 283 response.begin() 284 self.assertEqual(response.status, 408) 285 conn.close() 286 287 # Connect but send half the headers only. 288 self.persistent = True 289 conn = self.HTTP_CONN 290 conn.auto_open = False 291 conn.connect() 292 conn.send(ntob('GET /hello HTTP/1.1')) 293 conn.send(("Host: %s" % self.HOST).encode('ascii')) 294 295 # Wait for our socket timeout 296 time.sleep(timeout * 2) 297 298 # The conn should have already sent 408. 299 response = conn.response_class(conn.sock, method="GET") 300 response.begin() 301 self.assertEqual(response.status, 408) 302 conn.close()
303
305 # If we timeout after at least one request has succeeded, 306 # the server will close the conn without 408. 307 if cherrypy.server.protocol_version != "HTTP/1.1": 308 return self.skip() 309 310 self.PROTOCOL = "HTTP/1.1" 311 312 # Make an initial request 313 self.persistent = True 314 conn = self.HTTP_CONN 315 conn.putrequest("GET", "/timeout?t=%s" % timeout, skip_host=True) 316 conn.putheader("Host", self.HOST) 317 conn.endheaders() 318 response = conn.response_class(conn.sock, method="GET") 319 response.begin() 320 self.assertEqual(response.status, 200) 321 self.body = response.read() 322 self.assertBody(str(timeout)) 323 324 # Make a second request on the same socket 325 conn._output(ntob('GET /hello HTTP/1.1')) 326 conn._output(ntob("Host: %s" % self.HOST, 'ascii')) 327 conn._send_output() 328 response = conn.response_class(conn.sock, method="GET") 329 response.begin() 330 self.assertEqual(response.status, 200) 331 self.body = response.read() 332 self.assertBody("Hello, world!") 333 334 # Wait for our socket timeout 335 time.sleep(timeout * 2) 336 337 # Make another request on the same socket, which should error 338 conn._output(ntob('GET /hello HTTP/1.1')) 339 conn._output(ntob("Host: %s" % self.HOST, 'ascii')) 340 conn._send_output() 341 response = conn.response_class(conn.sock, method="GET") 342 try: 343 response.begin() 344 except: 345 if not isinstance(sys.exc_info()[1], 346 (socket.error, BadStatusLine)): 347 self.fail("Writing to timed out socket didn't fail" 348 " as it should have: %s" % sys.exc_info()[1]) 349 else: 350 if response.status != 408: 351 self.fail("Writing to timed out socket didn't fail" 352 " as it should have: %s" % 353 response.read()) 354 355 conn.close() 356 357 # Make another request on a new socket, which should work 358 self.persistent = True 359 conn = self.HTTP_CONN 360 conn.putrequest("GET", "/", skip_host=True) 361 conn.putheader("Host", self.HOST) 362 conn.endheaders() 363 response = conn.response_class(conn.sock, method="GET") 364 response.begin() 365 self.assertEqual(response.status, 200) 366 self.body = response.read() 367 self.assertBody(pov) 368 369 # Make another request on the same socket, 370 # but timeout on the headers 371 conn.send(ntob('GET /hello HTTP/1.1')) 372 # Wait for our socket timeout 373 time.sleep(timeout * 2) 374 response = conn.response_class(conn.sock, method="GET") 375 try: 376 response.begin() 377 except: 378 if not isinstance(sys.exc_info()[1], 379 (socket.error, BadStatusLine)): 380 self.fail("Writing to timed out socket didn't fail" 381 " as it should have: %s" % sys.exc_info()[1]) 382 else: 383 self.fail("Writing to timed out socket didn't fail" 384 " as it should have: %s" % 385 response.read()) 386 387 conn.close() 388 389 # Retry the request on a new connection, which should work 390 self.persistent = True 391 conn = self.HTTP_CONN 392 conn.putrequest("GET", "/", skip_host=True) 393 conn.putheader("Host", self.HOST) 394 conn.endheaders() 395 response = conn.response_class(conn.sock, method="GET") 396 response.begin() 397 self.assertEqual(response.status, 200) 398 self.body = response.read() 399 self.assertBody(pov) 400 conn.close()
401
402 - def test_HTTP11_pipelining(self):
403 if cherrypy.server.protocol_version != "HTTP/1.1": 404 return self.skip() 405 406 self.PROTOCOL = "HTTP/1.1" 407 408 # Test pipelining. httplib doesn't support this directly. 409 self.persistent = True 410 conn = self.HTTP_CONN 411 412 # Put request 1 413 conn.putrequest("GET", "/hello", skip_host=True) 414 conn.putheader("Host", self.HOST) 415 conn.endheaders() 416 417 for trial in range(5): 418 # Put next request 419 conn._output(ntob('GET /hello HTTP/1.1')) 420 conn._output(ntob("Host: %s" % self.HOST, 'ascii')) 421 conn._send_output() 422 423 # Retrieve previous response 424 response = conn.response_class(conn.sock, method="GET") 425 response.begin() 426 body = response.read(13) 427 self.assertEqual(response.status, 200) 428 self.assertEqual(body, ntob("Hello, world!")) 429 430 # Retrieve final response 431 response = conn.response_class(conn.sock, method="GET") 432 response.begin() 433 body = response.read() 434 self.assertEqual(response.status, 200) 435 self.assertEqual(body, ntob("Hello, world!")) 436 437 conn.close()
438
439 - def test_100_Continue(self):
440 if cherrypy.server.protocol_version != "HTTP/1.1": 441 return self.skip() 442 443 self.PROTOCOL = "HTTP/1.1" 444 445 self.persistent = True 446 conn = self.HTTP_CONN 447 448 # Try a page without an Expect request header first. 449 # Note that httplib's response.begin automatically ignores 450 # 100 Continue responses, so we must manually check for it. 451 conn.putrequest("POST", "/upload", skip_host=True) 452 conn.putheader("Host", self.HOST) 453 conn.putheader("Content-Type", "text/plain") 454 conn.putheader("Content-Length", "4") 455 conn.endheaders() 456 conn.send(ntob("d'oh")) 457 response = conn.response_class(conn.sock, method="POST") 458 version, status, reason = response._read_status() 459 self.assertNotEqual(status, 100) 460 conn.close() 461 462 # Now try a page with an Expect header... 463 conn.connect() 464 conn.putrequest("POST", "/upload", skip_host=True) 465 conn.putheader("Host", self.HOST) 466 conn.putheader("Content-Type", "text/plain") 467 conn.putheader("Content-Length", "17") 468 conn.putheader("Expect", "100-continue") 469 conn.endheaders() 470 response = conn.response_class(conn.sock, method="POST") 471 472 # ...assert and then skip the 100 response 473 version, status, reason = response._read_status() 474 self.assertEqual(status, 100) 475 while True: 476 line = response.fp.readline().strip() 477 if line: 478 self.fail( 479 "100 Continue should not output any headers. Got %r" % 480 line) 481 else: 482 break 483 484 # ...send the body 485 body = ntob("I am a small file") 486 conn.send(body) 487 488 # ...get the final response 489 response.begin() 490 self.status, self.headers, self.body = webtest.shb(response) 491 self.assertStatus(200) 492 self.assertBody("thanks for '%s'" % body) 493 conn.close()
494 495
496 -class ConnectionTests(helper.CPWebCase):
497 setup_server = staticmethod(setup_server) 498
499 - def test_readall_or_close(self):
500 if cherrypy.server.protocol_version != "HTTP/1.1": 501 return self.skip() 502 503 self.PROTOCOL = "HTTP/1.1" 504 505 if self.scheme == "https": 506 self.HTTP_CONN = HTTPSConnection 507 else: 508 self.HTTP_CONN = HTTPConnection 509 510 # Test a max of 0 (the default) and then reset to what it was above. 511 old_max = cherrypy.server.max_request_body_size 512 for new_max in (0, old_max): 513 cherrypy.server.max_request_body_size = new_max 514 515 self.persistent = True 516 conn = self.HTTP_CONN 517 518 # Get a POST page with an error 519 conn.putrequest("POST", "/err_before_read", skip_host=True) 520 conn.putheader("Host", self.HOST) 521 conn.putheader("Content-Type", "text/plain") 522 conn.putheader("Content-Length", "1000") 523 conn.putheader("Expect", "100-continue") 524 conn.endheaders() 525 response = conn.response_class(conn.sock, method="POST") 526 527 # ...assert and then skip the 100 response 528 version, status, reason = response._read_status() 529 self.assertEqual(status, 100) 530 while True: 531 skip = response.fp.readline().strip() 532 if not skip: 533 break 534 535 # ...send the body 536 conn.send(ntob("x" * 1000)) 537 538 # ...get the final response 539 response.begin() 540 self.status, self.headers, self.body = webtest.shb(response) 541 self.assertStatus(500) 542 543 # Now try a working page with an Expect header... 544 conn._output(ntob('POST /upload HTTP/1.1')) 545 conn._output(ntob("Host: %s" % self.HOST, 'ascii')) 546 conn._output(ntob("Content-Type: text/plain")) 547 conn._output(ntob("Content-Length: 17")) 548 conn._output(ntob("Expect: 100-continue")) 549 conn._send_output() 550 response = conn.response_class(conn.sock, method="POST") 551 552 # ...assert and then skip the 100 response 553 version, status, reason = response._read_status() 554 self.assertEqual(status, 100) 555 while True: 556 skip = response.fp.readline().strip() 557 if not skip: 558 break 559 560 # ...send the body 561 body = ntob("I am a small file") 562 conn.send(body) 563 564 # ...get the final response 565 response.begin() 566 self.status, self.headers, self.body = webtest.shb(response) 567 self.assertStatus(200) 568 self.assertBody("thanks for '%s'" % body) 569 conn.close()
570
571 - def test_No_Message_Body(self):
572 if cherrypy.server.protocol_version != "HTTP/1.1": 573 return self.skip() 574 575 self.PROTOCOL = "HTTP/1.1" 576 577 # Set our HTTP_CONN to an instance so it persists between requests. 578 self.persistent = True 579 580 # Make the first request and assert there's no "Connection: close". 581 self.getPage("/") 582 self.assertStatus('200 OK') 583 self.assertBody(pov) 584 self.assertNoHeader("Connection") 585 586 # Make a 204 request on the same connection. 587 self.getPage("/custom/204") 588 self.assertStatus(204) 589 self.assertNoHeader("Content-Length") 590 self.assertBody("") 591 self.assertNoHeader("Connection") 592 593 # Make a 304 request on the same connection. 594 self.getPage("/custom/304") 595 self.assertStatus(304) 596 self.assertNoHeader("Content-Length") 597 self.assertBody("") 598 self.assertNoHeader("Connection")
599
600 - def test_Chunked_Encoding(self):
601 if cherrypy.server.protocol_version != "HTTP/1.1": 602 return self.skip() 603 604 if (hasattr(self, 'harness') and 605 "modpython" in self.harness.__class__.__name__.lower()): 606 # mod_python forbids chunked encoding 607 return self.skip() 608 609 self.PROTOCOL = "HTTP/1.1" 610 611 # Set our HTTP_CONN to an instance so it persists between requests. 612 self.persistent = True 613 conn = self.HTTP_CONN 614 615 # Try a normal chunked request (with extensions) 616 body = ntob("8;key=value\r\nxx\r\nxxxx\r\n5\r\nyyyyy\r\n0\r\n" 617 "Content-Type: application/json\r\n" 618 "\r\n") 619 conn.putrequest("POST", "/upload", skip_host=True) 620 conn.putheader("Host", self.HOST) 621 conn.putheader("Transfer-Encoding", "chunked") 622 conn.putheader("Trailer", "Content-Type") 623 # Note that this is somewhat malformed: 624 # we shouldn't be sending Content-Length. 625 # RFC 2616 says the server should ignore it. 626 conn.putheader("Content-Length", "3") 627 conn.endheaders() 628 conn.send(body) 629 response = conn.getresponse() 630 self.status, self.headers, self.body = webtest.shb(response) 631 self.assertStatus('200 OK') 632 self.assertBody("thanks for '%s'" % ntob('xx\r\nxxxxyyyyy')) 633 634 # Try a chunked request that exceeds server.max_request_body_size. 635 # Note that the delimiters and trailer are included. 636 body = ntob("3e3\r\n" + ("x" * 995) + "\r\n0\r\n\r\n") 637 conn.putrequest("POST", "/upload", skip_host=True) 638 conn.putheader("Host", self.HOST) 639 conn.putheader("Transfer-Encoding", "chunked") 640 conn.putheader("Content-Type", "text/plain") 641 # Chunked requests don't need a content-length 642 ## conn.putheader("Content-Length", len(body)) 643 conn.endheaders() 644 conn.send(body) 645 response = conn.getresponse() 646 self.status, self.headers, self.body = webtest.shb(response) 647 self.assertStatus(413) 648 conn.close()
649
650 - def test_Content_Length_in(self):
651 # Try a non-chunked request where Content-Length exceeds 652 # server.max_request_body_size. Assert error before body send. 653 self.persistent = True 654 conn = self.HTTP_CONN 655 conn.putrequest("POST", "/upload", skip_host=True) 656 conn.putheader("Host", self.HOST) 657 conn.putheader("Content-Type", "text/plain") 658 conn.putheader("Content-Length", "9999") 659 conn.endheaders() 660 response = conn.getresponse() 661 self.status, self.headers, self.body = webtest.shb(response) 662 self.assertStatus(413) 663 self.assertBody("The entity sent with the request exceeds " 664 "the maximum allowed bytes.") 665 conn.close()
666
668 # Try a non-chunked response where Content-Length is less than 669 # the actual bytes in the response body. 670 self.persistent = True 671 conn = self.HTTP_CONN 672 conn.putrequest("GET", "/custom_cl?body=I+have+too+many+bytes&cl=5", 673 skip_host=True) 674 conn.putheader("Host", self.HOST) 675 conn.endheaders() 676 response = conn.getresponse() 677 self.status, self.headers, self.body = webtest.shb(response) 678 self.assertStatus(500) 679 self.assertBody( 680 "The requested resource returned more bytes than the " 681 "declared Content-Length.") 682 conn.close()
683
685 # Try a non-chunked response where Content-Length is less than 686 # the actual bytes in the response body. 687 self.persistent = True 688 conn = self.HTTP_CONN 689 conn.putrequest( 690 "GET", "/custom_cl?body=I+too&body=+have+too+many&cl=5", 691 skip_host=True) 692 conn.putheader("Host", self.HOST) 693 conn.endheaders() 694 response = conn.getresponse() 695 self.status, self.headers, self.body = webtest.shb(response) 696 self.assertStatus(200) 697 self.assertBody("I too") 698 conn.close()
699
700 - def test_598(self):
701 remote_data_conn = urlopen('%s://%s:%s/one_megabyte_of_a/' % 702 (self.scheme, self.HOST, self.PORT,)) 703 buf = remote_data_conn.read(512) 704 time.sleep(timeout * 0.6) 705 remaining = (1024 * 1024) - 512 706 while remaining: 707 data = remote_data_conn.read(remaining) 708 if not data: 709 break 710 else: 711 buf += data 712 remaining -= len(data) 713 714 self.assertEqual(len(buf), 1024 * 1024) 715 self.assertEqual(buf, ntob("a" * 1024 * 1024)) 716 self.assertEqual(remaining, 0) 717 remote_data_conn.close()
718 719
720 -class BadRequestTests(helper.CPWebCase):
721 setup_server = staticmethod(setup_server) 722
723 - def test_No_CRLF(self):
724 self.persistent = True 725 726 conn = self.HTTP_CONN 727 conn.send(ntob('GET /hello HTTP/1.1\n\n')) 728 response = conn.response_class(conn.sock, method="GET") 729 response.begin() 730 self.body = response.read() 731 self.assertBody("HTTP requires CRLF terminators") 732 conn.close() 733 734 conn.connect() 735 conn.send(ntob('GET /hello HTTP/1.1\r\n\n')) 736 response = conn.response_class(conn.sock, method="GET") 737 response.begin() 738 self.body = response.read() 739 self.assertBody("HTTP requires CRLF terminators") 740 conn.close()
741