Commit 27ba6401 authored by Vincent Pelletier's avatar Vincent Pelletier

test: More WSGI tests.

Also, drop redundant HTTP version fallback: this is already handled in
BaseHTTPRequestHandler.
parent d14ab1cb
......@@ -108,13 +108,10 @@ class CleanServerHandler(ServerHandler):
def setup_environ(self):
ServerHandler.setup_environ(self)
environ = self.environ
try:
request_major, request_minor = environ[
'SERVER_PROTOCOL'
].upper().split('/', 1)[1].split('.', 1)
request_version = (int(request_major), int(request_minor))
except (KeyError, ValueError):
request_version = (0, 9)
request_major, request_minor = environ[
'SERVER_PROTOCOL'
].upper().split('/', 1)[1].split('.', 1)
request_version = (int(request_major), int(request_minor))
if request_version > (1, 0):
if environ.get('HTTP_TRANSFER_ENCODING', '').lower() == 'chunked':
# XXX: does not support multiple encodings specified at once
......
......@@ -128,19 +128,24 @@ class FakeAppServer(object):
"""
return self._app
def noopApp(environ, start_response): # pylint: disable=unused-argument
class DummyApp(object):
"""
Minimal WSGI app, which does not touch environ.
Simple WSGI app with limited (but simple) scripting.
"""
start_response('200 OK', [])
return []
def __init__(self, callback=lambda x: []):
"""
callback (callable)
Received environ as argument,
Expected to return response body as an iterable.
"""
self._callback = callback
def readApp(environ, start_response):
"""
Minimal WSGI app, which reads request body.
"""
start_response('200 OK', [])
return [environ['wsgi.input'].read()]
def __call__(self, environ, start_response):
"""
WSGI entry point
"""
start_response('200 OK', [])
return self._callback(environ)
ON_EVENT_RAISE = object()
ON_EVENT_EXPIRE = object()
......@@ -1726,7 +1731,7 @@ class CaucaseTest(unittest.TestCase):
Not caucase-specific, but testing code which is part of caucase.
"""
def run(request_line_list, app=noopApp):
def run(request_line_list, app=DummyApp()):
"""
Trigger execution of app, with given request.
"""
......@@ -1771,9 +1776,10 @@ class CaucaseTest(unittest.TestCase):
getStatus(run(expect_continue_request)),
200,
)
read_app = DummyApp(lambda environ: [environ['wsgi.input'].read()])
# Read: 100 Continue (as first response, assume 200 OK happens later)
self.assertEqual(
getStatus(run(expect_continue_request, readApp)),
getStatus(run(expect_continue_request, read_app)),
100,
)
# HTTP/1.0: 200 OK
......@@ -1782,7 +1788,7 @@ class CaucaseTest(unittest.TestCase):
[
'PUT / HTTP/1.0',
] + expect_continue_request[1:],
readApp,
read_app,
)),
200,
)
......@@ -1799,9 +1805,74 @@ class CaucaseTest(unittest.TestCase):
'X-Chunked-Trailer: blah'
]
self.assertEqual(
getBody(run(chunked_request, readApp)),
getBody(run(chunked_request, read_app)),
'123456789abcd\r\nef0',
)
self.assertEqual(
getBody(run(
chunked_request,
DummyApp(lambda environ: [
environ['wsgi.input'].read(),
environ['wsgi.input'].read(),
]))),
'123456789abcd\r\nef0',
)
self.assertEqual(
getBody(run(
chunked_request,
DummyApp(lambda environ: [
environ['wsgi.input'].read(6),
environ['wsgi.input'].read(),
]))),
'123456789abcd\r\nef0',
)
self.assertEqual(
getBody(run(
chunked_request,
DummyApp(lambda environ: [
environ['wsgi.input'].read(32),
]))),
'123456789abcd\r\nef0',
)
self.assertEqual(
getStatus(run([
'PUT / HTTP/1.1',
'Transfer-Encoding: chunked',
'',
'1',
'abc', # Chunk longer than advertised in header.
], read_app)),
500,
)
self.assertEqual(
getStatus(run([
'PUT / HTTP/1.1',
'Transfer-Encoding: chunked',
'',
'y', # Not a valid chunk header
], read_app)),
500,
)
self.assertEqual(
getStatus(run([
'PUT / HTTP/1.1',
'Transfer-Encoding: chunked',
'',
'f;' + 'a' * 65537, # header too long
], read_app)),
500,
)
self.assertEqual(
getStatus(run([
'PUT / HTTP/1.1',
'Transfer-Encoding: chunked',
'',
'0',
'a' * 65537, # trailer too long
], read_app)),
500,
)
def testUpdater(self):
"""
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment