Commit 8af0b532 authored by Vincent Pelletier's avatar Vincent Pelletier

test: Add low-level WSGI tests.

parent 1a10a495
......@@ -1582,5 +1582,123 @@ class CaucaseTest(unittest.TestCase):
finally:
http.getBytePass = getBytePass_orig
def testWSGIBase(self):
"""
Low-level WSGI tests.
Not caucase-specific, but testing code which is part of caucase.
"""
def run(request_line_list, app=noopApp):
wfile = StringIO()
http.CaucaseWSGIRequestHandler(
FakeStreamRequest(
StringIO('\r\n'.join(request_line_list + [''])),
NoCloseFileProxy(wfile),
),
('0.0.0.0', 0),
FakeAppServer(app),
)
return wfile.getvalue().splitlines()
def getStatus(response_line_list):
version, code, reason = response_line_list[0].split(' ', 2)
return int(code)
def getBody(response_line_list):
return '\r\n'.join(response_line_list[response_line_list.index('') + 1:])
self.assertEqual(
getStatus(run(['GET /' + 'a' * 65537])),
414,
)
expect_continue_request = [
'PUT / HTTP/1.1',
'Expect: 100-continue',
'Content-Length: 4',
'Content-Type: text/plain',
'',
'Test',
]
# No read: 200 OK
self.assertEqual(
getStatus(run(expect_continue_request)),
200,
)
# Read: 100 Continue (as first response, assume 200 OK happens later)
self.assertEqual(
getStatus(run(expect_continue_request, readApp)),
100,
)
# HTTP/1.0: 200 OK
self.assertEqual(
getStatus(run(
[
'PUT / HTTP/1.0',
] + expect_continue_request[1:],
readApp,
)),
200,
)
chunked_request = [
'PUT / HTTP/1.1',
'Transfer-Encoding: chunked',
'',
'f;some=extension',
'123456789abcd\r\n',
'3',
'ef0',
'0',
'X-Chunked-Trailer: blah'
]
self.assertEqual(
getBody(run(chunked_request, readApp)),
'123456789abcd\r\nef0',
)
class FakeStreamRequest(object):
"""
For testing StreamRequestHandler subclasses
(like http.CaucaseWSGIRequestHandler).
"""
def __init__(self, rfile, wfile):
"""
rfile & wfile: Both halves of emulated socket.
"""
self._rfile = rfile
self._wfile = wfile
def makefile(self, mode, _):
"""
Fake splitting the socket.
"""
return self._rfile if 'r' in mode else self._wfile
class NoCloseFileProxy(object):
def __init__(self, real_file):
self._real_file = real_file
def __getattr__(self, name):
return getattr(self._real_file, name)
def close(self):
pass
class FakeAppServer(object):
def __init__(self, app):
self._app = app
self.base_environ = {}
def get_app(self):
return self._app
def noopApp(environ, start_response):
start_response('200 OK', [])
return []
def readApp(environ, start_response):
start_response('200 OK', [])
return [environ['wsgi.input'].read()]
if __name__ == '__main__':
unittest.main()
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