From 751093cfdde6021ba36477977540351a307398dd Mon Sep 17 00:00:00 2001 From: David Kerkeslager Date: Fri, 13 Dec 2019 17:15:53 -0500 Subject: [PATCH] Update tests, add subpath routing --- src/fwx/__init__.py | 46 ++++++++++- src/test_fwx.py | 189 ++++++++++++++++++++++++++++++++++++++++++++ test_fwx.py | 135 ------------------------------- 3 files changed, 232 insertions(+), 138 deletions(-) create mode 100644 src/test_fwx.py delete mode 100644 test_fwx.py diff --git a/src/fwx/__init__.py b/src/fwx/__init__.py index 411c077..c7d6442 100644 --- a/src/fwx/__init__.py +++ b/src/fwx/__init__.py @@ -24,7 +24,7 @@ _Request = collections.namedtuple( ) class Request(_Request): - def __new__(cls, env): + def __new__(cls, env, **kwargs): errors = [] accept = env.get('HTTP_ACCEPT') @@ -62,6 +62,10 @@ class Request(_Request): if method == 'GET': parameters = GET + elif method == 'POST': + raise Exception('not yet implemented') + else: + parameters = None result = super().__new__( cls, @@ -81,9 +85,16 @@ class Request(_Request): user_agent=user_agent, ) - result.subpath = path + if path.startswith('/'): + result.subpath = path[1:] + else: + result.subpath = path + return result +def _get_request_from_env(env): + return Request(env) + _Response = collections.namedtuple( 'Response', ( @@ -191,6 +202,35 @@ class RedirectResponse(_RedirectResponse): def content(self): return (b'',) +def default_file_not_found_handler(request): + return Response('', status=404) + +def route_on_subpath(**kwargs): + routes = kwargs.pop('routes') + file_not_found_handler = kwargs.pop( + 'file_not_found_hanlder', + default_file_not_found_handler, + ) + + if routes is None: + raise Exception('Keyword argument "routes" is required') + + if len(kwargs) > 0: + raise Exception('Unexpected keyword argument') + + def wrapped(request): + split_subpath = request.subpath.split('/', 1) + subpath = split_subpath[0] + + if len(split_subpath) == 2: + request.subpath = split_subpath[1] + else: + request.subpath = '' + + return routes.get(subpath, file_not_found_handler)(request) + + return wrapped + REQUEST_METHODS = ( 'GET', 'HEAD', @@ -258,7 +298,7 @@ def _get_content(response): def App(handler): def app(env, start_fn): - response = handler(Request(env)) + response = handler(_get_request_from_env(env)) start_fn(_get_status(response), _get_headers(response)) return _get_content(response) diff --git a/src/test_fwx.py b/src/test_fwx.py new file mode 100644 index 0000000..6f5eada --- /dev/null +++ b/src/test_fwx.py @@ -0,0 +1,189 @@ +import unittest +from unittest import mock + +import fwx + +class RequestTests(unittest.TestCase): + def test_GET(self): + request = fwx.Request({ + 'PATH_INFO': '/', + 'QUERY_STRING': 'foo=bar&baz=qux', + 'REQUEST_METHOD': 'GET', + }) + + self.assertEqual(request.GET['foo'], ['bar']) + self.assertEqual(request.GET['baz'], ['qux']) + + def test_parameters(self): + request = fwx.Request({ + 'PATH_INFO': '/', + 'REQUEST_METHOD': 'GET', + 'QUERY_STRING': 'foo=bar&baz=qux', + }) + + self.assertEqual(request.parameters['foo'], ['bar']) + self.assertEqual(request.parameters['baz'], ['qux']) + +class ResponseTests(unittest.TestCase): + def test_content_can_be_positional_argument(self): + response = fwx.Response('Hello, world\n', content_type='text/plain') + + self.assertEqual(response.content, 'Hello, world\n') + + def test_content_can_be_keyword_argument(self): + response = fwx.Response(content='Hello, world\n', content_type='text/plain') + + self.assertEqual(response.content, 'Hello, world\n') + + def test_status_defaults_to_200(self): + response = fwx.Response( + content_type='text/plain', + content='Hello, world\n', + ) + + self.assertEqual(response.status, 200) + + def test_headers(self): + response = fwx.Response( + content_type='text/plain', + content='Hello, world\n', + ) + + self.assertEqual( + response.headers, + ( + ('Content-Type', 'text/plain'), + ), + ) + +class HTMLResponseTests(unittest.TestCase): + def test_sets_content_type(self): + response = fwx.HTMLResponse('Hello, world') + self.assertEqual(response.content_type, 'text/html') + +class JSONResponseTests(unittest.TestCase): + def test_sets_content_type(self): + response = fwx.JSONResponse({ 'foo': 'bar', 'baz': 42 }) + self.assertEqual(response.content_type, 'application/json') + + def test_sets_content(self): + response = fwx.JSONResponse({ 'foo': 'bar', 'baz': 42 }) + self.assertEqual(response.content, '{"foo": "bar", "baz": 42}') + + def test_sets_content_json(self): + response = fwx.JSONResponse({ 'foo': 'bar', 'baz': 42 }) + self.assertEqual(response.content_json, {"foo": "bar", "baz": 42}) + +class TextResponseTests(unittest.TestCase): + def test_sets_content_type(self): + response = fwx.TextResponse('Hello, world\n') + self.assertEqual(response.content_type, 'text/plain') + +class RedirectResponse(unittest.TestCase): + def test_takes_location_as_positional_argument(self): + response = fwx.RedirectResponse('/location') + self.assertEqual(response.location, '/location') + + def test_takes_location_as_keyword_argument(self): + response = fwx.RedirectResponse(location='/location') + self.assertEqual(response.location, '/location') + + def test_permanent_defaults_to_true(self): + response = fwx.RedirectResponse('/location') + self.assertEqual(response.permanent, True) + + def test_status(self): + self.assertEqual( + fwx.RedirectResponse('/location', permanent=True).status, + 308, + ) + self.assertEqual( + fwx.RedirectResponse('/location', permanent=False).status, + 307, + ) + + def test_headers(self): + self.assertEqual( + fwx.RedirectResponse('/location').headers, + (('Location','/location'),), + ) + + def test_content(self): + self.assertEqual( + fwx.RedirectResponse('/location').content, + (b'',), + ) + +class _get_status_Tests(unittest.TestCase): + def test_basic(self): + self.assertEqual(fwx._get_status(mock.MagicMock(status=200)), '200 OK') + self.assertEqual(fwx._get_status(mock.MagicMock(status=307)), '307 Temporary Redirect') + self.assertEqual(fwx._get_status(mock.MagicMock(status=308)), '308 Permanent Redirect') + +class _get_content_Tests(unittest.TestCase): + def test_bytes(self): + self.assertEqual( + fwx._get_content(mock.MagicMock(content=b'Hello, world\n')), + (b'Hello, world\n',), + ) + + def test_str(self): + self.assertEqual( + fwx._get_content(mock.MagicMock(content='Hello, world\n')), + (b'Hello, world\n',), + ) + +class route_on_subpath_Tests(unittest.TestCase): + def test_routes(self): + router = fwx.route_on_subpath( + routes={ + 'foo': lambda request: fwx.TextResponse('foo'), + 'bar': lambda request: fwx.TextResponse('bar'), + 'baz': lambda request: fwx.TextResponse('baz'), + }, + ) + + self.assertEqual( + router(fwx.Request({ + 'PATH_INFO': '/bar/bara/anne/', + 'REQUEST_METHOD': 'GET', + })).content, + 'bar', + ) + + def test_resets_subpath(self): + router = fwx.route_on_subpath( + routes={ + 'foo': lambda request: fwx.TextResponse('foo'), + 'bar': lambda request: fwx.TextResponse(request.subpath), + 'baz': lambda request: fwx.TextResponse('baz'), + }, + ) + + self.assertEqual( + router(fwx.Request({ + 'PATH_INFO': '/bar/bara/anne/', + 'REQUEST_METHOD': 'GET', + })).content, + 'bara/anne/', + ) + + def test_leaves_path_intact(self): + router = fwx.route_on_subpath( + routes={ + 'foo': lambda request: fwx.TextResponse('foo'), + 'bar': lambda request: fwx.TextResponse(request.path), + 'baz': lambda request: fwx.TextResponse('baz'), + }, + ) + + self.assertEqual( + router(fwx.Request({ + 'PATH_INFO': '/bar/bara/anne/', + 'REQUEST_METHOD': 'GET', + })).content, + '/bar/bara/anne/', + ) + +if __name__ == '__main__': + unittest.main() diff --git a/test_fwx.py b/test_fwx.py deleted file mode 100644 index 672be96..0000000 --- a/test_fwx.py +++ /dev/null @@ -1,135 +0,0 @@ -import unittest -from unittest import mock - -import phial - -class RequestTests(unittest.TestCase): - def test_GET(self): - request = phial.Request({ - 'REQUEST_METHOD': 'GET', - 'QUERY_STRING': 'foo=bar&baz=qux', - }) - - self.assertEqual(request.GET['foo'], ['bar']) - self.assertEqual(request.GET['baz'], ['qux']) - - def test_parameters(self): - request = phial.Request({ - 'REQUEST_METHOD': 'GET', - 'QUERY_STRING': 'foo=bar&baz=qux', - }) - - self.assertEqual(request.parameters['foo'], ['bar']) - self.assertEqual(request.parameters['baz'], ['qux']) - -class ResponseTests(unittest.TestCase): - def test_content_can_be_positional_argument(self): - response = phial.Response('Hello, world\n', content_type='text/plain') - - self.assertEqual(response.content, 'Hello, world\n') - - def test_content_can_be_keyword_argument(self): - response = phial.Response(content='Hello, world\n', content_type='text/plain') - - self.assertEqual(response.content, 'Hello, world\n') - - def test_status_defaults_to_200(self): - response = phial.Response( - content_type='text/plain', - content='Hello, world\n', - ) - - self.assertEqual(response.status, 200) - - def test_headers(self): - response = phial.Response( - content_type='text/plain', - content='Hello, world\n', - ) - - self.assertEqual( - response.headers, - ( - ('Content-Type', 'text/plain'), - ), - ) - -class HTMLResponseTests(unittest.TestCase): - def test_sets_content_type(self): - response = phial.HTMLResponse('Hello, world') - self.assertEqual(response.content_type, 'text/html') - -class JSONResponseTests(unittest.TestCase): - def test_sets_content_type(self): - response = phial.JSONResponse({ 'foo': 'bar', 'baz': 42 }) - self.assertEqual(response.content_type, 'application/json') - - def test_sets_content(self): - response = phial.JSONResponse({ 'foo': 'bar', 'baz': 42 }) - self.assertEqual(response.content, '{"foo": "bar", "baz": 42}') - - def test_sets_content_json(self): - response = phial.JSONResponse({ 'foo': 'bar', 'baz': 42 }) - self.assertEqual(response.content_json, {"foo": "bar", "baz": 42}) - -class TextResponseTests(unittest.TestCase): - def test_sets_content_type(self): - response = phial.TextResponse('Hello, world\n') - self.assertEqual(response.content_type, 'text/plain') - -class RedirectResponse(unittest.TestCase): - def test_takes_location_as_positional_argument(self): - response = phial.RedirectResponse('/location') - self.assertEqual(response.location, '/location') - - def test_takes_location_as_keyword_argument(self): - response = phial.RedirectResponse(location='/location') - self.assertEqual(response.location, '/location') - - def test_permanent_defaults_to_true(self): - response = phial.RedirectResponse('/location') - self.assertEqual(response.permanent, True) - - def test_status(self): - self.assertEqual( - phial.RedirectResponse('/location', permanent=True).status, - 308, - ) - self.assertEqual( - phial.RedirectResponse('/location', permanent=False).status, - 307, - ) - - def test_headers(self): - self.assertEqual( - phial.RedirectResponse('/location').headers, - (('Location','/location'),), - ) - - def test_content(self): - self.assertEqual( - phial.RedirectResponse('/location').content, - (b'',), - ) - -class _get_status_Tests(unittest.TestCase): - def test_basic(self): - self.assertEqual(phial._get_status(mock.MagicMock(status=200)), '200 OK') - self.assertEqual(phial._get_status(mock.MagicMock(status=307)), '307 Temporary Redirect') - self.assertEqual(phial._get_status(mock.MagicMock(status=308)), '308 Permanent Redirect') - -class _get_content_Tests(unittest.TestCase): - def test_bytes(self): - self.assertEqual( - phial._get_content(mock.MagicMock(content=b'Hello, world\n')), - (b'Hello, world\n',), - ) - - def test_str(self): - self.assertEqual( - phial._get_content(mock.MagicMock(content='Hello, world\n')), - (b'Hello, world\n',), - ) - -if __name__ == '__main__': - unittest.main() -- 2.20.1