diff options
Diffstat (limited to 'tests/wsgi_test.py')
| -rw-r--r-- | tests/wsgi_test.py | 31 |
1 files changed, 21 insertions, 10 deletions
diff --git a/tests/wsgi_test.py b/tests/wsgi_test.py index a966038..5fcfa29 100644 --- a/tests/wsgi_test.py +++ b/tests/wsgi_test.py @@ -145,6 +145,17 @@ hello world """ +def recvall(socket_): + result = b'' + while True: + chunk = socket_.recv() + result += chunk + if chunk == b'': + break + + return result + + class ConnectionClosed(Exception): pass @@ -449,8 +460,8 @@ class TestHttpd(_TestBase): sock.write( b'POST /foo HTTP/1.1\r\nHost: localhost\r\n' b'Connection: close\r\nContent-length:3\r\n\r\nabc') - result = sock.read(8192) - self.assertEqual(result[-3:], b'abc') + result = recvall(sock) + assert result.endswith(b'abc') @tests.skip_if_no_ssl def test_013_empty_return(self): @@ -469,8 +480,8 @@ class TestHttpd(_TestBase): sock = eventlet.connect(('localhost', server_sock.getsockname()[1])) sock = eventlet.wrap_ssl(sock) sock.write(b'GET /foo HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n') - result = sock.read(8192) - self.assertEqual(result[-4:], b'\r\n\r\n') + result = recvall(sock) + assert result[-4:] == b'\r\n\r\n' def test_014_chunked_post(self): self.site.application = chunked_post @@ -964,9 +975,9 @@ class TestHttpd(_TestBase): try: client = ssl.wrap_socket(eventlet.connect(('localhost', port))) client.write(b'GET / HTTP/1.0\r\nHost: localhost\r\n\r\n') - result = client.read() - assert result.startswith('HTTP'), result - assert result.endswith('hello world') + result = recvall(client) + assert result.startswith(b'HTTP'), result + assert result.endswith(b'hello world') except ImportError: pass # TODO(openssl): should test with OpenSSL greenthread.kill(g) @@ -1163,8 +1174,8 @@ class TestHttpd(_TestBase): def test_unicode_raises_error(self): def wsgi_app(environ, start_response): start_response("200 OK", []) - yield u"oh hai" - yield u"non-encodable unicode: \u0230" + yield b"oh hai" + yield u"any unicode string should raise an error" self.site.application = wsgi_app sock = eventlet.connect(('localhost', self.port)) fd = sock.makefile('rwb') @@ -1173,7 +1184,7 @@ class TestHttpd(_TestBase): result = read_http(sock) self.assertEqual(result.status, 'HTTP/1.1 500 Internal Server Error') self.assertEqual(result.headers_lower['connection'], 'close') - assert b'unicode' in result.body + assert b'bytestring' in result.body def test_path_info_decoding(self): def wsgi_app(environ, start_response): |