diff options
Diffstat (limited to 'cherrypy/test/test_conn.py')
| -rw-r--r-- | cherrypy/test/test_conn.py | 352 |
1 files changed, 176 insertions, 176 deletions
diff --git a/cherrypy/test/test_conn.py b/cherrypy/test/test_conn.py index 1382c9e0..f7ce10cd 100644 --- a/cherrypy/test/test_conn.py +++ b/cherrypy/test/test_conn.py @@ -37,7 +37,7 @@ def setup_server(): @cherrypy.expose def hello(self): - return "Hello, world!" + return 'Hello, world!' @cherrypy.expose def timeout(self, t): @@ -69,16 +69,16 @@ def setup_server(): @cherrypy.expose def custom(self, response_code): cherrypy.response.status = response_code - return "Code = %s" % response_code + return 'Code = %s' % response_code @cherrypy.expose @cherrypy.config(**{'hooks.on_start_resource': raise500}) def err_before_read(self): - return "ok" + return 'ok' @cherrypy.expose def one_megabyte_of_a(self): - return ["a" * 1024] * 1024 + return ['a' * 1024] * 1024 @cherrypy.expose # Turn off the encoding tool so it doens't collapse @@ -106,33 +106,33 @@ class ConnectionCloseTests(helper.CPWebCase): setup_server = staticmethod(setup_server) def test_HTTP11(self): - if cherrypy.server.protocol_version != "HTTP/1.1": + if cherrypy.server.protocol_version != 'HTTP/1.1': return self.skip() - self.PROTOCOL = "HTTP/1.1" + self.PROTOCOL = 'HTTP/1.1' self.persistent = True # Make the first request and assert there's no "Connection: close". - self.getPage("/") + self.getPage('/') self.assertStatus('200 OK') self.assertBody(pov) - self.assertNoHeader("Connection") + self.assertNoHeader('Connection') # Make another request on the same connection. - self.getPage("/page1") + self.getPage('/page1') self.assertStatus('200 OK') self.assertBody(pov) - self.assertNoHeader("Connection") + self.assertNoHeader('Connection') # Test client-side close. - self.getPage("/page2", headers=[("Connection", "close")]) + self.getPage('/page2', headers=[('Connection', 'close')]) self.assertStatus('200 OK') self.assertBody(pov) - self.assertHeader("Connection", "close") + self.assertHeader('Connection', 'close') # Make another request on the same connection, which should error. - self.assertRaises(NotConnected, self.getPage, "/") + self.assertRaises(NotConnected, self.getPage, '/') def test_Streaming_no_len(self): try: @@ -153,25 +153,25 @@ class ConnectionCloseTests(helper.CPWebCase): pass def _streaming(self, set_cl): - if cherrypy.server.protocol_version == "HTTP/1.1": - self.PROTOCOL = "HTTP/1.1" + if cherrypy.server.protocol_version == 'HTTP/1.1': + self.PROTOCOL = 'HTTP/1.1' self.persistent = True # Make the first request and assert there's no "Connection: close". - self.getPage("/") + self.getPage('/') self.assertStatus('200 OK') self.assertBody(pov) - self.assertNoHeader("Connection") + self.assertNoHeader('Connection') # Make another, streamed request on the same connection. if set_cl: # When a Content-Length is provided, the content should stream # without closing the connection. - self.getPage("/stream?set_cl=Yes") - self.assertHeader("Content-Length") - self.assertNoHeader("Connection", "close") - self.assertNoHeader("Transfer-Encoding") + self.getPage('/stream?set_cl=Yes') + self.assertHeader('Content-Length') + self.assertNoHeader('Connection', 'close') + self.assertNoHeader('Transfer-Encoding') self.assertStatus('200 OK') self.assertBody('0123456789') @@ -179,78 +179,78 @@ class ConnectionCloseTests(helper.CPWebCase): # When no Content-Length response header is provided, # streamed output will either close the connection, or use # chunked encoding, to determine transfer-length. - self.getPage("/stream") - self.assertNoHeader("Content-Length") + self.getPage('/stream') + self.assertNoHeader('Content-Length') self.assertStatus('200 OK') self.assertBody('0123456789') chunked_response = False for k, v in self.headers: - if k.lower() == "transfer-encoding": - if str(v) == "chunked": + if k.lower() == 'transfer-encoding': + if str(v) == 'chunked': chunked_response = True if chunked_response: - self.assertNoHeader("Connection", "close") + self.assertNoHeader('Connection', 'close') else: - self.assertHeader("Connection", "close") + self.assertHeader('Connection', 'close') # Make another request on the same connection, which should # error. - self.assertRaises(NotConnected, self.getPage, "/") + self.assertRaises(NotConnected, self.getPage, '/') # Try HEAD. See # https://github.com/cherrypy/cherrypy/issues/864. - self.getPage("/stream", method='HEAD') + self.getPage('/stream', method='HEAD') self.assertStatus('200 OK') self.assertBody('') - self.assertNoHeader("Transfer-Encoding") + self.assertNoHeader('Transfer-Encoding') else: - self.PROTOCOL = "HTTP/1.0" + self.PROTOCOL = 'HTTP/1.0' self.persistent = True # Make the first request and assert Keep-Alive. - self.getPage("/", headers=[("Connection", "Keep-Alive")]) + self.getPage('/', headers=[('Connection', 'Keep-Alive')]) self.assertStatus('200 OK') self.assertBody(pov) - self.assertHeader("Connection", "Keep-Alive") + self.assertHeader('Connection', 'Keep-Alive') # Make another, streamed request on the same connection. if set_cl: # When a Content-Length is provided, the content should # stream without closing the connection. - self.getPage("/stream?set_cl=Yes", - headers=[("Connection", "Keep-Alive")]) - self.assertHeader("Content-Length") - self.assertHeader("Connection", "Keep-Alive") - self.assertNoHeader("Transfer-Encoding") + self.getPage('/stream?set_cl=Yes', + headers=[('Connection', 'Keep-Alive')]) + self.assertHeader('Content-Length') + self.assertHeader('Connection', 'Keep-Alive') + self.assertNoHeader('Transfer-Encoding') self.assertStatus('200 OK') self.assertBody('0123456789') else: # When a Content-Length is not provided, # the server should close the connection. - self.getPage("/stream", headers=[("Connection", "Keep-Alive")]) + self.getPage('/stream', headers=[('Connection', 'Keep-Alive')]) self.assertStatus('200 OK') self.assertBody('0123456789') - self.assertNoHeader("Content-Length") - self.assertNoHeader("Connection", "Keep-Alive") - self.assertNoHeader("Transfer-Encoding") + self.assertNoHeader('Content-Length') + self.assertNoHeader('Connection', 'Keep-Alive') + self.assertNoHeader('Transfer-Encoding') # Make another request on the same connection, which should # error. - self.assertRaises(NotConnected, self.getPage, "/") + self.assertRaises(NotConnected, self.getPage, '/') def test_HTTP10_KeepAlive(self): - self.PROTOCOL = "HTTP/1.0" - if self.scheme == "https": + self.PROTOCOL = 'HTTP/1.0' + if self.scheme == 'https': self.HTTP_CONN = HTTPSConnection else: self.HTTP_CONN = HTTPConnection # Test a normal HTTP/1.0 request. - self.getPage("/page2") + self.getPage('/page2') self.assertStatus('200 OK') self.assertBody(pov) # Apache, for example, may emit a Connection header even for HTTP/1.0 @@ -259,13 +259,13 @@ class ConnectionCloseTests(helper.CPWebCase): # Test a keep-alive HTTP/1.0 request. self.persistent = True - self.getPage("/page3", headers=[("Connection", "Keep-Alive")]) + self.getPage('/page3', headers=[('Connection', 'Keep-Alive')]) self.assertStatus('200 OK') self.assertBody(pov) - self.assertHeader("Connection", "Keep-Alive") + self.assertHeader('Connection', 'Keep-Alive') # Remove the keep-alive header again. - self.getPage("/page3") + self.getPage('/page3') self.assertStatus('200 OK') self.assertBody(pov) # Apache, for example, may emit a Connection header even for HTTP/1.0 @@ -278,10 +278,10 @@ class PipelineTests(helper.CPWebCase): def test_HTTP11_Timeout(self): # If we timeout without sending any data, # the server will close the conn with a 408. - if cherrypy.server.protocol_version != "HTTP/1.1": + if cherrypy.server.protocol_version != 'HTTP/1.1': return self.skip() - self.PROTOCOL = "HTTP/1.1" + self.PROTOCOL = 'HTTP/1.1' # Connect but send nothing. self.persistent = True @@ -293,7 +293,7 @@ class PipelineTests(helper.CPWebCase): time.sleep(timeout * 2) # The request should have returned 408 already. - response = conn.response_class(conn.sock, method="GET") + response = conn.response_class(conn.sock, method='GET') response.begin() self.assertEqual(response.status, 408) conn.close() @@ -304,13 +304,13 @@ class PipelineTests(helper.CPWebCase): conn.auto_open = False conn.connect() conn.send(ntob('GET /hello HTTP/1.1')) - conn.send(("Host: %s" % self.HOST).encode('ascii')) + conn.send(('Host: %s' % self.HOST).encode('ascii')) # Wait for our socket timeout time.sleep(timeout * 2) # The conn should have already sent 408. - response = conn.response_class(conn.sock, method="GET") + response = conn.response_class(conn.sock, method='GET') response.begin() self.assertEqual(response.status, 408) conn.close() @@ -318,18 +318,18 @@ class PipelineTests(helper.CPWebCase): def test_HTTP11_Timeout_after_request(self): # If we timeout after at least one request has succeeded, # the server will close the conn without 408. - if cherrypy.server.protocol_version != "HTTP/1.1": + if cherrypy.server.protocol_version != 'HTTP/1.1': return self.skip() - self.PROTOCOL = "HTTP/1.1" + self.PROTOCOL = 'HTTP/1.1' # Make an initial request self.persistent = True conn = self.HTTP_CONN - conn.putrequest("GET", "/timeout?t=%s" % timeout, skip_host=True) - conn.putheader("Host", self.HOST) + conn.putrequest('GET', '/timeout?t=%s' % timeout, skip_host=True) + conn.putheader('Host', self.HOST) conn.endheaders() - response = conn.response_class(conn.sock, method="GET") + response = conn.response_class(conn.sock, method='GET') response.begin() self.assertEqual(response.status, 200) self.body = response.read() @@ -337,33 +337,33 @@ class PipelineTests(helper.CPWebCase): # Make a second request on the same socket conn._output(ntob('GET /hello HTTP/1.1')) - conn._output(ntob("Host: %s" % self.HOST, 'ascii')) + conn._output(ntob('Host: %s' % self.HOST, 'ascii')) conn._send_output() - response = conn.response_class(conn.sock, method="GET") + response = conn.response_class(conn.sock, method='GET') response.begin() self.assertEqual(response.status, 200) self.body = response.read() - self.assertBody("Hello, world!") + self.assertBody('Hello, world!') # Wait for our socket timeout time.sleep(timeout * 2) # Make another request on the same socket, which should error conn._output(ntob('GET /hello HTTP/1.1')) - conn._output(ntob("Host: %s" % self.HOST, 'ascii')) + conn._output(ntob('Host: %s' % self.HOST, 'ascii')) conn._send_output() - response = conn.response_class(conn.sock, method="GET") + response = conn.response_class(conn.sock, method='GET') try: response.begin() except: if not isinstance(sys.exc_info()[1], (socket.error, BadStatusLine)): self.fail("Writing to timed out socket didn't fail" - " as it should have: %s" % sys.exc_info()[1]) + ' as it should have: %s' % sys.exc_info()[1]) else: if response.status != 408: self.fail("Writing to timed out socket didn't fail" - " as it should have: %s" % + ' as it should have: %s' % response.read()) conn.close() @@ -371,10 +371,10 @@ class PipelineTests(helper.CPWebCase): # Make another request on a new socket, which should work self.persistent = True conn = self.HTTP_CONN - conn.putrequest("GET", "/", skip_host=True) - conn.putheader("Host", self.HOST) + conn.putrequest('GET', '/', skip_host=True) + conn.putheader('Host', self.HOST) conn.endheaders() - response = conn.response_class(conn.sock, method="GET") + response = conn.response_class(conn.sock, method='GET') response.begin() self.assertEqual(response.status, 200) self.body = response.read() @@ -385,17 +385,17 @@ class PipelineTests(helper.CPWebCase): conn.send(ntob('GET /hello HTTP/1.1')) # Wait for our socket timeout time.sleep(timeout * 2) - response = conn.response_class(conn.sock, method="GET") + response = conn.response_class(conn.sock, method='GET') try: response.begin() except: if not isinstance(sys.exc_info()[1], (socket.error, BadStatusLine)): self.fail("Writing to timed out socket didn't fail" - " as it should have: %s" % sys.exc_info()[1]) + ' as it should have: %s' % sys.exc_info()[1]) else: self.fail("Writing to timed out socket didn't fail" - " as it should have: %s" % + ' as it should have: %s' % response.read()) conn.close() @@ -403,10 +403,10 @@ class PipelineTests(helper.CPWebCase): # Retry the request on a new connection, which should work self.persistent = True conn = self.HTTP_CONN - conn.putrequest("GET", "/", skip_host=True) - conn.putheader("Host", self.HOST) + conn.putrequest('GET', '/', skip_host=True) + conn.putheader('Host', self.HOST) conn.endheaders() - response = conn.response_class(conn.sock, method="GET") + response = conn.response_class(conn.sock, method='GET') response.begin() self.assertEqual(response.status, 200) self.body = response.read() @@ -414,53 +414,53 @@ class PipelineTests(helper.CPWebCase): conn.close() def test_HTTP11_pipelining(self): - if cherrypy.server.protocol_version != "HTTP/1.1": + if cherrypy.server.protocol_version != 'HTTP/1.1': return self.skip() - self.PROTOCOL = "HTTP/1.1" + self.PROTOCOL = 'HTTP/1.1' # Test pipelining. httplib doesn't support this directly. self.persistent = True conn = self.HTTP_CONN # Put request 1 - conn.putrequest("GET", "/hello", skip_host=True) - conn.putheader("Host", self.HOST) + conn.putrequest('GET', '/hello', skip_host=True) + conn.putheader('Host', self.HOST) conn.endheaders() for trial in range(5): # Put next request conn._output(ntob('GET /hello HTTP/1.1')) - conn._output(ntob("Host: %s" % self.HOST, 'ascii')) + conn._output(ntob('Host: %s' % self.HOST, 'ascii')) conn._send_output() # Retrieve previous response - response = conn.response_class(conn.sock, method="GET") + response = conn.response_class(conn.sock, method='GET') # there is a bug in python3 regarding the buffering of # ``conn.sock``. Until that bug get's fixed we will # monkey patch the ``reponse`` instance. # https://bugs.python.org/issue23377 if six.PY3: - response.fp = conn.sock.makefile("rb", 0) + response.fp = conn.sock.makefile('rb', 0) response.begin() body = response.read(13) self.assertEqual(response.status, 200) - self.assertEqual(body, ntob("Hello, world!")) + self.assertEqual(body, ntob('Hello, world!')) # Retrieve final response - response = conn.response_class(conn.sock, method="GET") + response = conn.response_class(conn.sock, method='GET') response.begin() body = response.read() self.assertEqual(response.status, 200) - self.assertEqual(body, ntob("Hello, world!")) + self.assertEqual(body, ntob('Hello, world!')) conn.close() def test_100_Continue(self): - if cherrypy.server.protocol_version != "HTTP/1.1": + if cherrypy.server.protocol_version != 'HTTP/1.1': return self.skip() - self.PROTOCOL = "HTTP/1.1" + self.PROTOCOL = 'HTTP/1.1' self.persistent = True conn = self.HTTP_CONN @@ -469,13 +469,13 @@ class PipelineTests(helper.CPWebCase): # Note that httplib's response.begin automatically ignores # 100 Continue responses, so we must manually check for it. try: - conn.putrequest("POST", "/upload", skip_host=True) - conn.putheader("Host", self.HOST) - conn.putheader("Content-Type", "text/plain") - conn.putheader("Content-Length", "4") + conn.putrequest('POST', '/upload', skip_host=True) + conn.putheader('Host', self.HOST) + conn.putheader('Content-Type', 'text/plain') + conn.putheader('Content-Length', '4') conn.endheaders() conn.send(ntob("d'oh")) - response = conn.response_class(conn.sock, method="POST") + response = conn.response_class(conn.sock, method='POST') version, status, reason = response._read_status() self.assertNotEqual(status, 100) finally: @@ -484,13 +484,13 @@ class PipelineTests(helper.CPWebCase): # Now try a page with an Expect header... try: conn.connect() - conn.putrequest("POST", "/upload", skip_host=True) - conn.putheader("Host", self.HOST) - conn.putheader("Content-Type", "text/plain") - conn.putheader("Content-Length", "17") - conn.putheader("Expect", "100-continue") + conn.putrequest('POST', '/upload', skip_host=True) + conn.putheader('Host', self.HOST) + conn.putheader('Content-Type', 'text/plain') + conn.putheader('Content-Length', '17') + conn.putheader('Expect', '100-continue') conn.endheaders() - response = conn.response_class(conn.sock, method="POST") + response = conn.response_class(conn.sock, method='POST') # ...assert and then skip the 100 response version, status, reason = response._read_status() @@ -499,13 +499,13 @@ class PipelineTests(helper.CPWebCase): line = response.fp.readline().strip() if line: self.fail( - "100 Continue should not output any headers. Got %r" % + '100 Continue should not output any headers. Got %r' % line) else: break # ...send the body - body = ntob("I am a small file") + body = ntob('I am a small file') conn.send(body) # ...get the final response @@ -521,12 +521,12 @@ class ConnectionTests(helper.CPWebCase): setup_server = staticmethod(setup_server) def test_readall_or_close(self): - if cherrypy.server.protocol_version != "HTTP/1.1": + if cherrypy.server.protocol_version != 'HTTP/1.1': return self.skip() - self.PROTOCOL = "HTTP/1.1" + self.PROTOCOL = 'HTTP/1.1' - if self.scheme == "https": + if self.scheme == 'https': self.HTTP_CONN = HTTPSConnection else: self.HTTP_CONN = HTTPConnection @@ -540,13 +540,13 @@ class ConnectionTests(helper.CPWebCase): conn = self.HTTP_CONN # Get a POST page with an error - conn.putrequest("POST", "/err_before_read", skip_host=True) - conn.putheader("Host", self.HOST) - conn.putheader("Content-Type", "text/plain") - conn.putheader("Content-Length", "1000") - conn.putheader("Expect", "100-continue") + conn.putrequest('POST', '/err_before_read', skip_host=True) + conn.putheader('Host', self.HOST) + conn.putheader('Content-Type', 'text/plain') + conn.putheader('Content-Length', '1000') + conn.putheader('Expect', '100-continue') conn.endheaders() - response = conn.response_class(conn.sock, method="POST") + response = conn.response_class(conn.sock, method='POST') # ...assert and then skip the 100 response version, status, reason = response._read_status() @@ -557,7 +557,7 @@ class ConnectionTests(helper.CPWebCase): break # ...send the body - conn.send(ntob("x" * 1000)) + conn.send(ntob('x' * 1000)) # ...get the final response response.begin() @@ -566,12 +566,12 @@ class ConnectionTests(helper.CPWebCase): # Now try a working page with an Expect header... conn._output(ntob('POST /upload HTTP/1.1')) - conn._output(ntob("Host: %s" % self.HOST, 'ascii')) - conn._output(ntob("Content-Type: text/plain")) - conn._output(ntob("Content-Length: 17")) - conn._output(ntob("Expect: 100-continue")) + conn._output(ntob('Host: %s' % self.HOST, 'ascii')) + conn._output(ntob('Content-Type: text/plain')) + conn._output(ntob('Content-Length: 17')) + conn._output(ntob('Expect: 100-continue')) conn._send_output() - response = conn.response_class(conn.sock, method="POST") + response = conn.response_class(conn.sock, method='POST') # ...assert and then skip the 100 response version, status, reason = response._read_status() @@ -582,7 +582,7 @@ class ConnectionTests(helper.CPWebCase): break # ...send the body - body = ntob("I am a small file") + body = ntob('I am a small file') conn.send(body) # ...get the final response @@ -593,61 +593,61 @@ class ConnectionTests(helper.CPWebCase): conn.close() def test_No_Message_Body(self): - if cherrypy.server.protocol_version != "HTTP/1.1": + if cherrypy.server.protocol_version != 'HTTP/1.1': return self.skip() - self.PROTOCOL = "HTTP/1.1" + self.PROTOCOL = 'HTTP/1.1' # Set our HTTP_CONN to an instance so it persists between requests. self.persistent = True # Make the first request and assert there's no "Connection: close". - self.getPage("/") + self.getPage('/') self.assertStatus('200 OK') self.assertBody(pov) - self.assertNoHeader("Connection") + self.assertNoHeader('Connection') # Make a 204 request on the same connection. - self.getPage("/custom/204") + self.getPage('/custom/204') self.assertStatus(204) - self.assertNoHeader("Content-Length") - self.assertBody("") - self.assertNoHeader("Connection") + self.assertNoHeader('Content-Length') + self.assertBody('') + self.assertNoHeader('Connection') # Make a 304 request on the same connection. - self.getPage("/custom/304") + self.getPage('/custom/304') self.assertStatus(304) - self.assertNoHeader("Content-Length") - self.assertBody("") - self.assertNoHeader("Connection") + self.assertNoHeader('Content-Length') + self.assertBody('') + self.assertNoHeader('Connection') def test_Chunked_Encoding(self): - if cherrypy.server.protocol_version != "HTTP/1.1": + if cherrypy.server.protocol_version != 'HTTP/1.1': return self.skip() if (hasattr(self, 'harness') and - "modpython" in self.harness.__class__.__name__.lower()): + 'modpython' in self.harness.__class__.__name__.lower()): # mod_python forbids chunked encoding return self.skip() - self.PROTOCOL = "HTTP/1.1" + self.PROTOCOL = 'HTTP/1.1' # Set our HTTP_CONN to an instance so it persists between requests. self.persistent = True conn = self.HTTP_CONN # Try a normal chunked request (with extensions) - body = ntob("8;key=value\r\nxx\r\nxxxx\r\n5\r\nyyyyy\r\n0\r\n" - "Content-Type: application/json\r\n" - "\r\n") - conn.putrequest("POST", "/upload", skip_host=True) - conn.putheader("Host", self.HOST) - conn.putheader("Transfer-Encoding", "chunked") - conn.putheader("Trailer", "Content-Type") + body = ntob('8;key=value\r\nxx\r\nxxxx\r\n5\r\nyyyyy\r\n0\r\n' + 'Content-Type: application/json\r\n' + '\r\n') + conn.putrequest('POST', '/upload', skip_host=True) + conn.putheader('Host', self.HOST) + conn.putheader('Transfer-Encoding', 'chunked') + conn.putheader('Trailer', 'Content-Type') # Note that this is somewhat malformed: # we shouldn't be sending Content-Length. # RFC 2616 says the server should ignore it. - conn.putheader("Content-Length", "3") + conn.putheader('Content-Length', '3') conn.endheaders() conn.send(body) response = conn.getresponse() @@ -657,11 +657,11 @@ class ConnectionTests(helper.CPWebCase): # Try a chunked request that exceeds server.max_request_body_size. # Note that the delimiters and trailer are included. - body = ntob("3e3\r\n" + ("x" * 995) + "\r\n0\r\n\r\n") - conn.putrequest("POST", "/upload", skip_host=True) - conn.putheader("Host", self.HOST) - conn.putheader("Transfer-Encoding", "chunked") - conn.putheader("Content-Type", "text/plain") + body = ntob('3e3\r\n' + ('x' * 995) + '\r\n0\r\n\r\n') + conn.putrequest('POST', '/upload', skip_host=True) + conn.putheader('Host', self.HOST) + conn.putheader('Transfer-Encoding', 'chunked') + conn.putheader('Content-Type', 'text/plain') # Chunked requests don't need a content-length ## conn.putheader("Content-Length", len(body)) conn.endheaders() @@ -676,16 +676,16 @@ class ConnectionTests(helper.CPWebCase): # server.max_request_body_size. Assert error before body send. self.persistent = True conn = self.HTTP_CONN - conn.putrequest("POST", "/upload", skip_host=True) - conn.putheader("Host", self.HOST) - conn.putheader("Content-Type", "text/plain") - conn.putheader("Content-Length", "9999") + conn.putrequest('POST', '/upload', skip_host=True) + conn.putheader('Host', self.HOST) + conn.putheader('Content-Type', 'text/plain') + conn.putheader('Content-Length', '9999') conn.endheaders() response = conn.getresponse() self.status, self.headers, self.body = webtest.shb(response) self.assertStatus(413) - self.assertBody("The entity sent with the request exceeds " - "the maximum allowed bytes.") + self.assertBody('The entity sent with the request exceeds ' + 'the maximum allowed bytes.') conn.close() def test_Content_Length_out_preheaders(self): @@ -693,16 +693,16 @@ class ConnectionTests(helper.CPWebCase): # the actual bytes in the response body. self.persistent = True conn = self.HTTP_CONN - conn.putrequest("GET", "/custom_cl?body=I+have+too+many+bytes&cl=5", + conn.putrequest('GET', '/custom_cl?body=I+have+too+many+bytes&cl=5', skip_host=True) - conn.putheader("Host", self.HOST) + conn.putheader('Host', self.HOST) conn.endheaders() response = conn.getresponse() self.status, self.headers, self.body = webtest.shb(response) self.assertStatus(500) self.assertBody( - "The requested resource returned more bytes than the " - "declared Content-Length.") + 'The requested resource returned more bytes than the ' + 'declared Content-Length.') conn.close() def test_Content_Length_out_postheaders(self): @@ -711,14 +711,14 @@ class ConnectionTests(helper.CPWebCase): self.persistent = True conn = self.HTTP_CONN conn.putrequest( - "GET", "/custom_cl?body=I+too&body=+have+too+many&cl=5", + 'GET', '/custom_cl?body=I+too&body=+have+too+many&cl=5', skip_host=True) - conn.putheader("Host", self.HOST) + conn.putheader('Host', self.HOST) conn.endheaders() response = conn.getresponse() self.status, self.headers, self.body = webtest.shb(response) self.assertStatus(200) - self.assertBody("I too") + self.assertBody('I too') conn.close() def test_598(self): @@ -736,7 +736,7 @@ class ConnectionTests(helper.CPWebCase): remaining -= len(data) self.assertEqual(len(buf), 1024 * 1024) - self.assertEqual(buf, ntob("a" * 1024 * 1024)) + self.assertEqual(buf, ntob('a' * 1024 * 1024)) self.assertEqual(remaining, 0) remote_data_conn.close() @@ -765,12 +765,12 @@ socket_reset_errors = [ for name in reset_names if hasattr(errno, name) ] -"reset error numbers available on this platform" +'reset error numbers available on this platform' socket_reset_errors += [ # Python 3.5 raises an http.client.RemoteDisconnected # with this message - "Remote end closed connection without response", + 'Remote end closed connection without response', ] @@ -786,10 +786,10 @@ class LimitedRequestQueueTests(helper.CPWebCase): # all of wsgiserver's WorkerThreads and fill its Queue. for i in range(15): conn = self.HTTP_CONN(self.HOST, self.PORT) - conn.putrequest("POST", "/upload", skip_host=True) - conn.putheader("Host", self.HOST) - conn.putheader("Content-Type", "text/plain") - conn.putheader("Content-Length", "4") + conn.putrequest('POST', '/upload', skip_host=True) + conn.putheader('Host', self.HOST) + conn.putheader('Content-Type', 'text/plain') + conn.putheader('Content-Length', '4') conn.endheaders() conns.append(conn) @@ -804,10 +804,10 @@ class LimitedRequestQueueTests(helper.CPWebCase): overflow_conn.sock.connect(sa) break - overflow_conn.putrequest("GET", "/", skip_host=True) - overflow_conn.putheader("Host", self.HOST) + overflow_conn.putrequest('GET', '/', skip_host=True) + overflow_conn.putheader('Host', self.HOST) overflow_conn.endheaders() - response = overflow_conn.response_class(overflow_conn.sock, method="GET") + response = overflow_conn.response_class(overflow_conn.sock, method='GET') try: response.begin() except socket.error as exc: @@ -815,8 +815,8 @@ class LimitedRequestQueueTests(helper.CPWebCase): pass # Expected. else: tmpl = ( - "Overflow conn did not get RST. " - "Got {exc.args!r} instead" + 'Overflow conn did not get RST. ' + 'Got {exc.args!r} instead' ) raise AssertionError(tmpl.format(**locals())) except BadStatusLine: @@ -824,11 +824,11 @@ class LimitedRequestQueueTests(helper.CPWebCase): # RST correctly. assert sys.platform == 'darwin' else: - raise AssertionError("Overflow conn did not get RST ") + raise AssertionError('Overflow conn did not get RST ') finally: for conn in conns: - conn.send(ntob("done")) - response = conn.response_class(conn.sock, method="POST") + conn.send(ntob('done')) + response = conn.response_class(conn.sock, method='POST') response.begin() self.body = response.read() self.assertBody("thanks for 'done'") @@ -845,16 +845,16 @@ class BadRequestTests(helper.CPWebCase): conn = self.HTTP_CONN conn.send(ntob('GET /hello HTTP/1.1\n\n')) - response = conn.response_class(conn.sock, method="GET") + response = conn.response_class(conn.sock, method='GET') response.begin() self.body = response.read() - self.assertBody("HTTP requires CRLF terminators") + self.assertBody('HTTP requires CRLF terminators') conn.close() conn.connect() conn.send(ntob('GET /hello HTTP/1.1\r\n\n')) - response = conn.response_class(conn.sock, method="GET") + response = conn.response_class(conn.sock, method='GET') response.begin() self.body = response.read() - self.assertBody("HTTP requires CRLF terminators") + self.assertBody('HTTP requires CRLF terminators') conn.close() |