summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorMichael Merickel <michael@merickel.org>2019-07-19 01:01:46 -0500
committerMichael Merickel <michael@merickel.org>2019-07-19 01:01:46 -0500
commit9cf94669a9bb21d51741f5840b765e4ab5f94d48 (patch)
tree8fa81217a2b0e18c01f4ea4c7b7d407605501297
parenta15155f54e8ff615fd3f1db3cc35a60127fb58f3 (diff)
downloadwaitress-9cf94669a9bb21d51741f5840b765e4ab5f94d48.tar.gz
support echoing more request info, not just body, in functional tests
-rw-r--r--waitress/tests/fixtureapps/echo.py46
-rw-r--r--waitress/tests/test_functional.py45
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):