Commit f2805289 authored by Vincent Pelletier's avatar Vincent Pelletier

test: Make pylint happier.

parent 124dda69
......@@ -76,6 +76,72 @@ def retry(callback, try_count=200, try_delay=0.1):
return result
time.sleep(try_delay)
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):
"""
Intercept .close() calls, for example to allow reading StringIO content
despite attempted closure.
"""
def __init__(self, real_file):
"""
real_file: Object toforward all non-close calls to.
"""
self._real_file = real_file
def __getattr__(self, name):
return getattr(self._real_file, name)
def close(self):
"""
Don't.
"""
pass
class FakeAppServer(object):
"""
Minimal fake WSGI app server.
"""
def __init__(self, app):
self._app = app
self.base_environ = {}
def get_app(self):
"""
Retrieve app.
"""
return self._app
def noopApp(environ, start_response):
"""
Minimal WSGI app, which does not touch environ.
"""
start_response('200 OK', [])
return []
def readApp(environ, start_response):
"""
Minimal WSGI app, which reads request body.
"""
start_response('200 OK', [])
return [environ['wsgi.input'].read()]
class CaucaseTest(unittest.TestCase):
"""
Test a complete caucase setup: spawn a caucase-http server on CAUCASE_NETLOC
......@@ -99,7 +165,7 @@ class CaucaseTest(unittest.TestCase):
self._client_user_crl = os.path.join(client_dir, 'cau.crl.pem')
# pylint: enable=bad-whitespace
self._server_event = server_event = threading.Event()
self._server_event = threading.Event()
self._server_dir = server_dir = os.path.join(data_dir, 'server')
os.mkdir(server_dir)
# pylint: disable=bad-whitespace
......@@ -1589,6 +1655,9 @@ class CaucaseTest(unittest.TestCase):
Not caucase-specific, but testing code which is part of caucase.
"""
def run(request_line_list, app=noopApp):
"""
Trigger execution of app, with given request.
"""
wfile = StringIO()
http.CaucaseWSGIRequestHandler(
FakeStreamRequest(
......@@ -1601,10 +1670,16 @@ class CaucaseTest(unittest.TestCase):
return wfile.getvalue().splitlines()
def getStatus(response_line_list):
version, code, reason = response_line_list[0].split(' ', 2)
"""
Naive extraction of http status out of an http response.
"""
_, code, _ = response_line_list[0].split(' ', 2)
return int(code)
def getBody(response_line_list):
"""
Naive extraction of http response body.
"""
return '\r\n'.join(response_line_list[response_line_list.index('') + 1:])
self.assertEqual(
......@@ -1656,49 +1731,5 @@ class CaucaseTest(unittest.TestCase):
'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