diff options
author | Michael Merickel <michael@merickel.org> | 2019-07-19 01:01:46 -0500 |
---|---|---|
committer | Michael Merickel <michael@merickel.org> | 2019-07-19 01:01:46 -0500 |
commit | 9cf94669a9bb21d51741f5840b765e4ab5f94d48 (patch) | |
tree | 8fa81217a2b0e18c01f4ea4c7b7d407605501297 | |
parent | a15155f54e8ff615fd3f1db3cc35a60127fb58f3 (diff) | |
download | waitress-9cf94669a9bb21d51741f5840b765e4ab5f94d48.tar.gz |
support echoing more request info, not just body, in functional tests
-rw-r--r-- | waitress/tests/fixtureapps/echo.py | 46 | ||||
-rw-r--r-- | waitress/tests/test_functional.py | 45 |
2 files changed, 64 insertions, 27 deletions
diff --git a/waitress/tests/fixtureapps/echo.py b/waitress/tests/fixtureapps/echo.py index f5fd5d1..8a01ea8 100644 --- a/waitress/tests/fixtureapps/echo.py +++ b/waitress/tests/fixtureapps/echo.py @@ -1,11 +1,43 @@ +from collections import namedtuple +import json + def app(environ, start_response): # pragma: no cover cl = environ.get('CONTENT_LENGTH', None) if cl is not None: cl = int(cl) - body = environ['wsgi.input'].read(cl) - cl = str(len(body)) - start_response( - '200 OK', - [('Content-Length', cl), ('Content-Type', 'text/plain')] - ) - return [body] + request_body = environ['wsgi.input'].read(cl) + cl = str(len(request_body)) + meta = { + 'method': environ['REQUEST_METHOD'], + 'path_info': environ['PATH_INFO'], + 'script_name': environ['SCRIPT_NAME'], + 'query_string': environ['QUERY_STRING'], + 'content_length': cl, + 'scheme': environ['wsgi.url_scheme'], + 'remote_addr': environ['REMOTE_ADDR'], + 'remote_host': environ['REMOTE_HOST'], + 'server_port': environ['SERVER_PORT'], + 'server_name': environ['SERVER_NAME'], + 'headers': { + k[len('HTTP_'):]: v + for k, v in environ.items() + if k.startswith('HTTP_') + }, + } + response = json.dumps(meta).encode('utf8') + b'\r\n\r\n' + request_body + start_response('200 OK', [ + ('Content-Length', str(len(response))), + ('Content-Type', 'text/plain'), + ]) + return [response] + + +Echo = namedtuple('Echo', ( + 'method path_info script_name query_string content_length scheme ' + 'remote_addr remote_host server_port server_name headers body' +)) + +def parse_response(response): + meta, body = response.split(b'\r\n\r\n', 1) + meta = json.loads(meta.decode('utf8')) + return Echo(body=body, **meta) diff --git a/waitress/tests/test_functional.py b/waitress/tests/test_functional.py index f3170b9..6748c5f 100644 --- a/waitress/tests/test_functional.py +++ b/waitress/tests/test_functional.py @@ -144,6 +144,11 @@ class EchoTests(object): def tearDown(self): self.stop_subprocess() + def _read_echo(self, fp): + from waitress.tests.fixtureapps import echo + line, headers, response_body = read_http(fp) + return line, headers, echo.parse_response(response_body) + def test_date_and_server(self): to_send = ("GET / HTTP/1.0\n" "Content-Length: 0\n\n") @@ -151,7 +156,7 @@ class EchoTests(object): self.connect() self.sock.send(to_send) fp = self.sock.makefile('rb', 0) - line, headers, response_body = read_http(fp) + line, headers, echo = self._read_echo(fp) self.assertline(line, '200', 'OK', 'HTTP/1.0') self.assertEqual(headers.get('server'), 'waitress') self.assertTrue(headers.get('date')) @@ -177,10 +182,10 @@ class EchoTests(object): self.connect() self.sock.send(to_send) fp = self.sock.makefile('rb', 0) - line, headers, response_body = read_http(fp) + line, headers, echo = self._read_echo(fp) self.assertline(line, '200', 'OK', 'HTTP/1.0') - self.assertEqual(headers.get('content-length'), '5') - self.assertEqual(response_body, b'hello') + self.assertEqual(echo.content_length, '5') + self.assertEqual(echo.body, b'hello') def test_send_empty_body(self): to_send = ("GET / HTTP/1.0\n" @@ -189,10 +194,10 @@ class EchoTests(object): self.connect() self.sock.send(to_send) fp = self.sock.makefile('rb', 0) - line, headers, response_body = read_http(fp) + line, headers, echo = self._read_echo(fp) self.assertline(line, '200', 'OK', 'HTTP/1.0') - self.assertEqual(headers.get('content-length'), '0') - self.assertEqual(response_body, b'') + self.assertEqual(echo.content_length, '0') + self.assertEqual(echo.body, b'') def test_multiple_requests_with_body(self): orig_sock = self.sock @@ -222,11 +227,11 @@ class EchoTests(object): self.connect() self.sock.send(s) fp = self.sock.makefile('rb', 0) - line, headers, response_body = read_http(fp) + line, headers, echo = self._read_echo(fp) self.assertline(line, '200', 'OK', 'HTTP/1.0') - self.assertEqual(int(headers['content-length']), len(data)) - self.assertEqual(len(response_body), len(data)) - self.assertEqual(response_body, tobytes(data)) + self.assertEqual(int(echo.content_length), len(data)) + self.assertEqual(len(echo.body), len(data)) + self.assertEqual(echo.body, tobytes(data)) def test_large_body(self): # 1024 characters. @@ -240,10 +245,10 @@ class EchoTests(object): self.connect() self.sock.send(s) fp = self.sock.makefile('rb', 0) - line, headers, response_body = read_http(fp) + line, headers, echo = self._read_echo(fp) self.assertline(line, '200', 'OK', 'HTTP/1.0') - self.assertEqual(headers.get('content-length'), '1024') - self.assertEqual(response_body, tobytes(body)) + self.assertEqual(echo.content_length, '1024') + self.assertEqual(echo.body, tobytes(body)) def test_many_clients(self): conns = [] @@ -270,10 +275,10 @@ class EchoTests(object): self.sock.send(header) self.sock.send(b"0\r\n\r\n") fp = self.sock.makefile('rb', 0) - line, headers, response_body = read_http(fp) + line, headers, echo = self._read_echo(fp) self.assertline(line, '200', 'OK', 'HTTP/1.1') - self.assertEqual(response_body, b'') - self.assertEqual(headers['content-length'], '0') + self.assertEqual(echo.body, b'') + self.assertEqual(echo.content_length, '0') self.assertFalse('transfer-encoding' in headers) def test_chunking_request_with_content(self): @@ -291,10 +296,10 @@ class EchoTests(object): self.sock.send(control_line) self.sock.send(s) self.sock.send(b"0\r\n\r\n") - line, headers, response_body = read_http(fp) + line, headers, echo = self._read_echo(fp) self.assertline(line, '200', 'OK', 'HTTP/1.1') - self.assertEqual(response_body, expected) - self.assertEqual(headers['content-length'], str(len(expected))) + self.assertEqual(echo.body, expected) + self.assertEqual(echo.content_length, str(len(expected))) self.assertFalse('transfer-encoding' in headers) def test_broken_chunked_encoding(self): |