Async test client
This commit is contained in:
Miguel Grinberg
@@ -138,7 +138,7 @@ class Response(BaseResponse):
|
||||
# body
|
||||
try:
|
||||
async for body in self.body_iter():
|
||||
if isinstance(body, str):
|
||||
if isinstance(body, str): # pragma: no cover
|
||||
body = body.encode()
|
||||
await stream.awrite(body)
|
||||
except OSError as exc: # pragma: no cover
|
||||
@@ -340,12 +340,21 @@ class Microdot(BaseMicrodot):
|
||||
res = await self._invoke_handler(
|
||||
f, req, **req.url_args)
|
||||
if isinstance(res, tuple):
|
||||
res = Response(*res)
|
||||
body = res[0]
|
||||
if isinstance(res[1], int):
|
||||
status_code = res[1]
|
||||
headers = res[2] if len(res) > 2 else {}
|
||||
else:
|
||||
status_code = 200
|
||||
headers = res[1]
|
||||
res = Response(body, status_code, headers)
|
||||
elif not isinstance(res, Response):
|
||||
res = Response(res)
|
||||
for handler in self.after_request_handlers:
|
||||
res = await self._invoke_handler(
|
||||
handler, req, res) or res
|
||||
for handler in req.after_request_handlers:
|
||||
res = await handler(req, res) or res
|
||||
elif f in self.error_handlers:
|
||||
res = await self._invoke_handler(
|
||||
self.error_handlers[f], req)
|
||||
|
||||
126
src/microdot_asyncio_test_client.py
Normal file
126
src/microdot_asyncio_test_client.py
Normal file
@@ -0,0 +1,126 @@
|
||||
from microdot_asyncio import Request, _AsyncBytesIO
|
||||
from microdot_test_client import TestClient as BaseTestClient, \
|
||||
TestResponse as BaseTestResponse
|
||||
|
||||
|
||||
class TestResponse(BaseTestResponse):
|
||||
"""A response object issued by the Microdot test client."""
|
||||
|
||||
@classmethod
|
||||
async def create(cls, res):
|
||||
test_res = cls()
|
||||
test_res._initialize_response(res)
|
||||
await test_res._initialize_body(res)
|
||||
test_res._process_text_body()
|
||||
test_res._process_json_body()
|
||||
return test_res
|
||||
|
||||
async def _initialize_body(self, res):
|
||||
self.body = b''
|
||||
async for body in res.body_iter():
|
||||
if isinstance(body, str):
|
||||
body = body.encode()
|
||||
self.body += body
|
||||
|
||||
|
||||
class TestClient(BaseTestClient):
|
||||
"""A test client for Microdot's Asynchronous web server.
|
||||
|
||||
:param app: The Microdot application instance.
|
||||
:param cookies: A dictionary of cookies to use when sending requests to the
|
||||
application.
|
||||
|
||||
The following example shows how to create a test client for an application
|
||||
and send a test request::
|
||||
|
||||
from microdot_asyncio import Microdot
|
||||
|
||||
app = Microdot()
|
||||
|
||||
@app.get('/')
|
||||
async def index():
|
||||
return 'Hello, World!'
|
||||
|
||||
async def test_hello_world(self):
|
||||
client = TestClient(app)
|
||||
res = await client.get('/')
|
||||
assert res.status_code == 200
|
||||
assert res.text == 'Hello, World!'
|
||||
"""
|
||||
async def request(self, method, path, headers=None, body=None):
|
||||
headers = headers or {}
|
||||
body, headers = self._process_body(body, headers)
|
||||
cookies, headers = self._process_cookies(headers)
|
||||
request_bytes = self._render_request(method, path, headers, body)
|
||||
|
||||
req = await Request.create(self.app, _AsyncBytesIO(request_bytes),
|
||||
('127.0.0.1', 1234))
|
||||
res = await self.app.dispatch_request(req)
|
||||
res.complete()
|
||||
|
||||
self._update_cookies(res)
|
||||
return await TestResponse.create(res)
|
||||
|
||||
async def get(self, path, headers=None):
|
||||
"""Send a GET request to the application.
|
||||
|
||||
:param path: The request URL.
|
||||
:param headers: A dictionary of headers to send with the request.
|
||||
|
||||
This method returns a
|
||||
:class:`TestResponse <microdot_test_client.TestResponse>` object.
|
||||
"""
|
||||
return await self.request('GET', path, headers=headers)
|
||||
|
||||
async def post(self, path, headers=None, body=None):
|
||||
"""Send a POST request to the application.
|
||||
|
||||
:param path: The request URL.
|
||||
:param headers: A dictionary of headers to send with the request.
|
||||
:param body: The request body. If a dictionary or list is provided,
|
||||
a JSON-encoded body will be sent. A string body is encoded
|
||||
to bytes as UTF-8. A bytes body is sent as-is.
|
||||
|
||||
This method returns a
|
||||
:class:`TestResponse <microdot_test_client.TestResponse>` object.
|
||||
"""
|
||||
return await self.request('POST', path, headers=headers, body=body)
|
||||
|
||||
async def put(self, path, headers=None, body=None):
|
||||
"""Send a PUT request to the application.
|
||||
|
||||
:param path: The request URL.
|
||||
:param headers: A dictionary of headers to send with the request.
|
||||
:param body: The request body. If a dictionary or list is provided,
|
||||
a JSON-encoded body will be sent. A string body is encoded
|
||||
to bytes as UTF-8. A bytes body is sent as-is.
|
||||
|
||||
This method returns a
|
||||
:class:`TestResponse <microdot_test_client.TestResponse>` object.
|
||||
"""
|
||||
return await self.request('PUT', path, headers=headers, body=body)
|
||||
|
||||
async def patch(self, path, headers=None, body=None):
|
||||
"""Send a PATCH request to the application.
|
||||
|
||||
:param path: The request URL.
|
||||
:param headers: A dictionary of headers to send with the request.
|
||||
:param body: The request body. If a dictionary or list is provided,
|
||||
a JSON-encoded body will be sent. A string body is encoded
|
||||
to bytes as UTF-8. A bytes body is sent as-is.
|
||||
|
||||
This method returns a
|
||||
:class:`TestResponse <microdot_test_client.TestResponse>` object.
|
||||
"""
|
||||
return await self.request('PATCH', path, headers=headers, body=body)
|
||||
|
||||
async def delete(self, path, headers=None):
|
||||
"""Send a DELETE request to the application.
|
||||
|
||||
:param path: The request URL.
|
||||
:param headers: A dictionary of headers to send with the request.
|
||||
|
||||
This method returns a
|
||||
:class:`TestResponse <microdot_test_client.TestResponse>` object.
|
||||
"""
|
||||
return await self.request('DELETE', path, headers=headers)
|
||||
@@ -5,35 +5,58 @@ from microdot import Request
|
||||
|
||||
class TestResponse:
|
||||
"""A response object issued by the Microdot test client."""
|
||||
def __init__(self, res):
|
||||
def __init__(self):
|
||||
#: The numeric status code returned by the server.
|
||||
self.status_code = res.status_code
|
||||
self.status_code = None
|
||||
#: The text reason associated with the status response, such as
|
||||
#: ``'OK'`` or ``'NOT FOUND'``.
|
||||
self.reason = res.reason
|
||||
#: ``'OK'`` or ``'NOT FOUND'``. Set to ``None`` unless the application
|
||||
#: explicitly sets it on the response object.
|
||||
self.reason = None
|
||||
#: A dictionary with the response headers.
|
||||
self.headers = res.headers
|
||||
self.headers = None
|
||||
#: The body of the response, as a bytes object.
|
||||
self.body = None
|
||||
#: The body of the response, decoded to a UTF-8 string. Set to
|
||||
#: ``None`` if the response cannot be represented as UTF-8 text.
|
||||
self.text = None
|
||||
#: The body of the JSON response, decoded to a dictionary or list. Set
|
||||
#: ``Note`` if the response does not have a JSON payload.
|
||||
self.json = None
|
||||
|
||||
def _initialize_response(self, res):
|
||||
self.status_code = res.status_code
|
||||
self.reason = res.reason
|
||||
self.headers = res.headers
|
||||
|
||||
def _initialize_body(self, res):
|
||||
self.body = b''
|
||||
for body in res.body_iter():
|
||||
if isinstance(body, str):
|
||||
body = body.encode()
|
||||
self.body += body
|
||||
|
||||
def _process_text_body(self):
|
||||
try:
|
||||
#: The body of the response, decoded to a UTF-8 string. Set to
|
||||
#: ``None`` if the response cannot be represented as UTF-8 text.
|
||||
self.text = self.body.decode()
|
||||
except ValueError:
|
||||
self.text = None
|
||||
#: The body of the JSON response, decoded to a dictionary or list. Set
|
||||
#: ``Note`` if the response does not have a JSON payload.
|
||||
self.json = None
|
||||
pass
|
||||
|
||||
def _process_json_body(self):
|
||||
for name, value in self.headers.items(): # pragma: no branch
|
||||
if name.lower() == 'content-type':
|
||||
if value.lower() == 'application/json':
|
||||
self.json = json.loads(self.text)
|
||||
break
|
||||
|
||||
@classmethod
|
||||
def create(cls, res):
|
||||
test_res = cls()
|
||||
test_res._initialize_response(res)
|
||||
test_res._initialize_body(res)
|
||||
test_res._process_text_body()
|
||||
test_res._process_json_body()
|
||||
return test_res
|
||||
|
||||
|
||||
class TestClient:
|
||||
"""A test client for Microdot.
|
||||
@@ -63,9 +86,7 @@ class TestClient:
|
||||
self.app = app
|
||||
self.cookies = cookies or {}
|
||||
|
||||
def request(self, method, path, headers=None, body=None):
|
||||
if headers is None: # pragma: no branch
|
||||
headers = {}
|
||||
def _process_body(self, body, headers):
|
||||
if body is None:
|
||||
body = b''
|
||||
elif isinstance(body, (dict, list)):
|
||||
@@ -78,6 +99,11 @@ class TestClient:
|
||||
if body and 'Content-Length' not in headers and \
|
||||
'content-length' not in headers:
|
||||
headers['Content-Length'] = str(len(body))
|
||||
if 'Host' not in headers: # pragma: no branch
|
||||
headers['Host'] = 'example.com:1234'
|
||||
return body, headers
|
||||
|
||||
def _process_cookies(self, headers):
|
||||
cookies = ''
|
||||
for name, value in self.cookies.items():
|
||||
if cookies:
|
||||
@@ -88,20 +114,18 @@ class TestClient:
|
||||
headers['Cookie'] += '; ' + cookies
|
||||
else:
|
||||
headers['Cookie'] = cookies
|
||||
return cookies, headers
|
||||
|
||||
def _render_request(self, method, path, headers, body):
|
||||
request_bytes = '{method} {path} HTTP/1.0\n'.format(
|
||||
method=method, path=path)
|
||||
if 'Host' not in headers: # pragma: no branch
|
||||
headers['Host'] = 'example.com:1234'
|
||||
for header, value in headers.items():
|
||||
request_bytes += '{header}: {value}\n'.format(
|
||||
header=header, value=value)
|
||||
request_bytes = request_bytes.encode() + b'\n' + body
|
||||
return request_bytes
|
||||
|
||||
req = Request.create(self.app, BytesIO(request_bytes),
|
||||
('127.0.0.1', 1234))
|
||||
res = self.app.dispatch_request(req)
|
||||
res.complete()
|
||||
|
||||
def _update_cookies(self, res):
|
||||
for name, value in res.headers.items():
|
||||
if name.lower() == 'set-cookie':
|
||||
for cookie in value:
|
||||
@@ -122,7 +146,20 @@ class TestClient:
|
||||
del self.cookies[cookie_name]
|
||||
else:
|
||||
self.cookies[cookie_name] = cookie_options[0]
|
||||
return TestResponse(res)
|
||||
|
||||
def request(self, method, path, headers=None, body=None):
|
||||
headers = headers or {}
|
||||
body, headers = self._process_body(body, headers)
|
||||
cookies, headers = self._process_cookies(headers)
|
||||
request_bytes = self._render_request(method, path, headers, body)
|
||||
|
||||
req = Request.create(self.app, BytesIO(request_bytes),
|
||||
('127.0.0.1', 1234))
|
||||
res = self.app.dispatch_request(req)
|
||||
res.complete()
|
||||
|
||||
self._update_cookies(res)
|
||||
return TestResponse.create(res)
|
||||
|
||||
def get(self, path, headers=None):
|
||||
"""Send a GET request to the application.
|
||||
|
||||
@@ -5,18 +5,17 @@ from microdot_test_client import TestClient
|
||||
from tests import mock_socket
|
||||
|
||||
|
||||
def mock_create_thread(f, *args, **kwargs):
|
||||
f(*args, **kwargs)
|
||||
|
||||
|
||||
class TestMicrodot(unittest.TestCase):
|
||||
def _mock_socket(self):
|
||||
def _mock(self):
|
||||
def mock_create_thread(f, *args, **kwargs):
|
||||
f(*args, **kwargs)
|
||||
|
||||
self.original_socket = sys.modules['microdot'].socket
|
||||
self.original_create_thread = sys.modules['microdot'].create_thread
|
||||
sys.modules['microdot'].socket = mock_socket
|
||||
sys.modules['microdot'].create_thread = mock_create_thread
|
||||
|
||||
def _unmock_socket(self):
|
||||
def _unmock(self):
|
||||
sys.modules['microdot'].socket = self.original_socket
|
||||
sys.modules['microdot'].create_thread = self.original_create_thread
|
||||
|
||||
@@ -39,6 +38,7 @@ class TestMicrodot(unittest.TestCase):
|
||||
res = client.get('/')
|
||||
self.assertEqual(res.status_code, 200)
|
||||
self.assertEqual(res.headers['Content-Type'], 'text/plain')
|
||||
self.assertEqual(res.headers['Content-Length'], '3')
|
||||
self.assertEqual(res.text, 'foo')
|
||||
self.assertEqual(res.body, b'foo')
|
||||
self.assertEqual(res.json, None)
|
||||
@@ -62,7 +62,7 @@ class TestMicrodot(unittest.TestCase):
|
||||
self.assertEqual(res.text, 'bar')
|
||||
|
||||
def test_empty_request(self):
|
||||
self._mock_socket()
|
||||
self._mock()
|
||||
|
||||
app = Microdot()
|
||||
|
||||
@@ -76,7 +76,7 @@ class TestMicrodot(unittest.TestCase):
|
||||
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
|
||||
self.assertTrue(fd.response.endswith(b'\r\n\r\nBad request'))
|
||||
|
||||
self._unmock_socket()
|
||||
self._unmock()
|
||||
|
||||
def test_method_decorators(self):
|
||||
app = Microdot()
|
||||
@@ -265,7 +265,7 @@ class TestMicrodot(unittest.TestCase):
|
||||
self.assertEqual(res.text, 'baz')
|
||||
|
||||
def test_400(self):
|
||||
self._mock_socket()
|
||||
self._mock()
|
||||
|
||||
app = Microdot()
|
||||
|
||||
@@ -279,10 +279,10 @@ class TestMicrodot(unittest.TestCase):
|
||||
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
|
||||
self.assertTrue(fd.response.endswith(b'\r\n\r\nBad request'))
|
||||
|
||||
self._unmock_socket()
|
||||
self._unmock()
|
||||
|
||||
def test_400_handler(self):
|
||||
self._mock_socket()
|
||||
self._mock()
|
||||
|
||||
app = Microdot()
|
||||
|
||||
@@ -300,7 +300,7 @@ class TestMicrodot(unittest.TestCase):
|
||||
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
|
||||
self.assertTrue(fd.response.endswith(b'\r\n\r\n400'))
|
||||
|
||||
self._unmock_socket()
|
||||
self._unmock()
|
||||
|
||||
def test_404(self):
|
||||
app = Microdot()
|
||||
|
||||
@@ -1,17 +1,30 @@
|
||||
try:
|
||||
import uasyncio as asyncio
|
||||
except ImportError:
|
||||
import asyncio
|
||||
import sys
|
||||
import unittest
|
||||
from microdot_asyncio import Microdot, Response
|
||||
from microdot_asyncio_test_client import TestClient
|
||||
from tests import mock_asyncio, mock_socket
|
||||
|
||||
|
||||
class TestMicrodotAsync(unittest.TestCase):
|
||||
def setUp(self):
|
||||
# mock socket module
|
||||
self._mock()
|
||||
|
||||
def tearDown(self):
|
||||
self._unmock()
|
||||
|
||||
def _run(self, coro):
|
||||
loop = asyncio.get_event_loop()
|
||||
return loop.run_until_complete(coro)
|
||||
|
||||
def _mock(self):
|
||||
self.original_asyncio = sys.modules['microdot_asyncio'].asyncio
|
||||
sys.modules['microdot_asyncio'].asyncio = mock_asyncio
|
||||
|
||||
def tearDown(self):
|
||||
# restore original socket module
|
||||
def _unmock(self):
|
||||
sys.modules['microdot_asyncio'].asyncio = self.original_asyncio
|
||||
|
||||
def _add_shutdown(self, app):
|
||||
@@ -33,19 +46,23 @@ class TestMicrodotAsync(unittest.TestCase):
|
||||
async def index2(req):
|
||||
return 'foo-async'
|
||||
|
||||
mock_socket.clear_requests()
|
||||
fd = mock_socket.add_request('GET', '/')
|
||||
fd2 = mock_socket.add_request('GET', '/async')
|
||||
self._add_shutdown(app)
|
||||
app.run()
|
||||
self.assertTrue(fd.response.startswith(b'HTTP/1.0 200 OK\r\n'))
|
||||
self.assertIn(b'Content-Length: 3\r\n', fd.response)
|
||||
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
|
||||
self.assertTrue(fd.response.endswith(b'\r\n\r\nfoo'))
|
||||
self.assertTrue(fd2.response.startswith(b'HTTP/1.0 200 OK\r\n'))
|
||||
self.assertIn(b'Content-Length: 9\r\n', fd2.response)
|
||||
self.assertIn(b'Content-Type: text/plain\r\n', fd2.response)
|
||||
self.assertTrue(fd2.response.endswith(b'\r\n\r\nfoo-async'))
|
||||
client = TestClient(app)
|
||||
|
||||
res = self._run(client.get('/'))
|
||||
self.assertEqual(res.status_code, 200)
|
||||
self.assertEqual(res.headers['Content-Type'], 'text/plain')
|
||||
self.assertEqual(res.headers['Content-Length'], '3')
|
||||
self.assertEqual(res.text, 'foo')
|
||||
self.assertEqual(res.body, b'foo')
|
||||
self.assertEqual(res.json, None)
|
||||
|
||||
res = self._run(client.get('/async'))
|
||||
self.assertEqual(res.status_code, 200)
|
||||
self.assertEqual(res.headers['Content-Type'], 'text/plain')
|
||||
self.assertEqual(res.headers['Content-Length'], '9')
|
||||
self.assertEqual(res.text, 'foo-async')
|
||||
self.assertEqual(res.body, b'foo-async')
|
||||
self.assertEqual(res.json, None)
|
||||
|
||||
def test_post_request(self):
|
||||
app = Microdot()
|
||||
@@ -62,19 +79,23 @@ class TestMicrodotAsync(unittest.TestCase):
|
||||
async def index_post2(req):
|
||||
return Response('bar-async')
|
||||
|
||||
mock_socket.clear_requests()
|
||||
fd = mock_socket.add_request('POST', '/')
|
||||
fd2 = mock_socket.add_request('POST', '/async')
|
||||
self._add_shutdown(app)
|
||||
app.run()
|
||||
self.assertTrue(fd.response.startswith(b'HTTP/1.0 200 OK\r\n'))
|
||||
self.assertIn(b'Content-Length: 3\r\n', fd.response)
|
||||
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
|
||||
self.assertTrue(fd.response.endswith(b'\r\n\r\nbar'))
|
||||
self.assertTrue(fd2.response.startswith(b'HTTP/1.0 200 OK\r\n'))
|
||||
self.assertIn(b'Content-Length: 9\r\n', fd2.response)
|
||||
self.assertIn(b'Content-Type: text/plain\r\n', fd2.response)
|
||||
self.assertTrue(fd2.response.endswith(b'\r\n\r\nbar-async'))
|
||||
client = TestClient(app)
|
||||
|
||||
res = self._run(client.post('/'))
|
||||
self.assertEqual(res.status_code, 200)
|
||||
self.assertEqual(res.headers['Content-Type'], 'text/plain')
|
||||
self.assertEqual(res.headers['Content-Length'], '3')
|
||||
self.assertEqual(res.text, 'bar')
|
||||
self.assertEqual(res.body, b'bar')
|
||||
self.assertEqual(res.json, None)
|
||||
|
||||
res = self._run(client.post('/async'))
|
||||
self.assertEqual(res.status_code, 200)
|
||||
self.assertEqual(res.headers['Content-Type'], 'text/plain')
|
||||
self.assertEqual(res.headers['Content-Length'], '9')
|
||||
self.assertEqual(res.text, 'bar-async')
|
||||
self.assertEqual(res.body, b'bar-async')
|
||||
self.assertEqual(res.json, None)
|
||||
|
||||
def test_empty_request(self):
|
||||
app = Microdot()
|
||||
@@ -89,12 +110,153 @@ class TestMicrodotAsync(unittest.TestCase):
|
||||
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
|
||||
self.assertTrue(fd.response.endswith(b'\r\n\r\nBad request'))
|
||||
|
||||
def test_method_decorators(self):
|
||||
app = Microdot()
|
||||
|
||||
@app.get('/get')
|
||||
def get(req):
|
||||
return 'GET'
|
||||
|
||||
@app.post('/post')
|
||||
async def post(req):
|
||||
return 'POST'
|
||||
|
||||
@app.put('/put')
|
||||
def put(req):
|
||||
return 'PUT'
|
||||
|
||||
@app.patch('/patch')
|
||||
async def patch(req):
|
||||
return 'PATCH'
|
||||
|
||||
@app.delete('/delete')
|
||||
def delete(req):
|
||||
return 'DELETE'
|
||||
|
||||
client = TestClient(app)
|
||||
methods = ['GET', 'POST', 'PUT', 'PATCH', 'DELETE']
|
||||
for method in methods:
|
||||
res = self._run(getattr(
|
||||
client, method.lower())('/' + method.lower()))
|
||||
self.assertEqual(res.status_code, 200)
|
||||
self.assertEqual(res.headers['Content-Type'], 'text/plain')
|
||||
self.assertEqual(res.text, method)
|
||||
|
||||
def test_headers(self):
|
||||
app = Microdot()
|
||||
|
||||
@app.route('/')
|
||||
def index(req):
|
||||
return req.headers.get('X-Foo')
|
||||
|
||||
client = TestClient(app)
|
||||
res = self._run(client.get('/', headers={'X-Foo': 'bar'}))
|
||||
self.assertEqual(res.status_code, 200)
|
||||
self.assertEqual(res.headers['Content-Type'], 'text/plain')
|
||||
self.assertEqual(res.text, 'bar')
|
||||
|
||||
def test_cookies(self):
|
||||
app = Microdot()
|
||||
|
||||
@app.route('/')
|
||||
def index(req):
|
||||
return req.cookies['one'] + req.cookies['two'] + \
|
||||
req.cookies['three']
|
||||
|
||||
client = TestClient(app, cookies={'one': '1', 'two': '2'})
|
||||
res = self._run(client.get('/', headers={'Cookie': 'three=3'}))
|
||||
self.assertEqual(res.status_code, 200)
|
||||
self.assertEqual(res.headers['Content-Type'], 'text/plain')
|
||||
self.assertEqual(res.text, '123')
|
||||
|
||||
def test_binary_payload(self):
|
||||
app = Microdot()
|
||||
|
||||
@app.post('/')
|
||||
def index(req):
|
||||
return req.body
|
||||
|
||||
client = TestClient(app)
|
||||
res = self._run(client.post('/', body=b'foo'))
|
||||
self.assertEqual(res.status_code, 200)
|
||||
self.assertEqual(res.headers['Content-Type'], 'text/plain')
|
||||
self.assertEqual(res.text, 'foo')
|
||||
|
||||
def test_json_payload(self):
|
||||
app = Microdot()
|
||||
|
||||
@app.post('/dict')
|
||||
def json_dict(req):
|
||||
print(req.headers)
|
||||
return req.json.get('foo')
|
||||
|
||||
@app.post('/list')
|
||||
def json_list(req):
|
||||
return req.json[0]
|
||||
|
||||
client = TestClient(app)
|
||||
|
||||
res = self._run(client.post('/dict', body={'foo': 'bar'}))
|
||||
self.assertEqual(res.status_code, 200)
|
||||
self.assertEqual(res.headers['Content-Type'], 'text/plain')
|
||||
self.assertEqual(res.text, 'bar')
|
||||
|
||||
res = self._run(client.post('/list', body=['foo', 'bar']))
|
||||
self.assertEqual(res.status_code, 200)
|
||||
self.assertEqual(res.headers['Content-Type'], 'text/plain')
|
||||
self.assertEqual(res.text, 'foo')
|
||||
|
||||
def test_tuple_responses(self):
|
||||
app = Microdot()
|
||||
|
||||
@app.route('/body')
|
||||
def one(req):
|
||||
return 'one'
|
||||
|
||||
@app.route('/body-status')
|
||||
def two(req):
|
||||
return 'two', 202
|
||||
|
||||
@app.route('/body-headers')
|
||||
def three(req):
|
||||
return '<p>three</p>', {'Content-Type': 'text/html'}
|
||||
|
||||
@app.route('/body-status-headers')
|
||||
def four(req):
|
||||
return '<p>four</p>', 202, {'Content-Type': 'text/html'}
|
||||
|
||||
client = TestClient(app)
|
||||
|
||||
res = self._run(client.get('/body'))
|
||||
self.assertEqual(res.status_code, 200)
|
||||
self.assertEqual(res.headers['Content-Type'], 'text/plain')
|
||||
self.assertEqual(res.text, 'one')
|
||||
|
||||
res = self._run(client.get('/body-status'))
|
||||
self.assertEqual(res.status_code, 202)
|
||||
self.assertEqual(res.headers['Content-Type'], 'text/plain')
|
||||
self.assertEqual(res.text, 'two')
|
||||
|
||||
res = self._run(client.get('/body-headers'))
|
||||
self.assertEqual(res.status_code, 200)
|
||||
self.assertEqual(res.headers['Content-Type'], 'text/html')
|
||||
self.assertEqual(res.text, '<p>three</p>')
|
||||
|
||||
res = self._run(client.get('/body-status-headers'))
|
||||
self.assertEqual(res.status_code, 202)
|
||||
self.assertEqual(res.headers['Content-Type'], 'text/html')
|
||||
self.assertEqual(res.text, '<p>four</p>')
|
||||
|
||||
def test_before_after_request(self):
|
||||
app = Microdot()
|
||||
|
||||
@app.before_request
|
||||
def before_request(req):
|
||||
if req.path == '/bar':
|
||||
@req.after_request
|
||||
async def after_request(req, res):
|
||||
res.headers['X-Two'] = '2'
|
||||
return res
|
||||
return 'bar', 202
|
||||
req.g.message = 'baz'
|
||||
|
||||
@@ -115,29 +277,29 @@ class TestMicrodotAsync(unittest.TestCase):
|
||||
def baz(req):
|
||||
return req.g.message
|
||||
|
||||
mock_socket.clear_requests()
|
||||
fd = mock_socket.add_request('GET', '/bar')
|
||||
self._add_shutdown(app)
|
||||
app.run()
|
||||
self.assertTrue(fd.response.startswith(b'HTTP/1.0 202 N/A\r\n'))
|
||||
self.assertIn(b'X-One: 1\r\n', fd.response)
|
||||
self.assertIn(b'Set-Cookie: foo=bar\r\n', fd.response)
|
||||
self.assertIn(b'Content-Length: 3\r\n', fd.response)
|
||||
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
|
||||
self.assertTrue(fd.response.endswith(b'\r\n\r\nbar'))
|
||||
client = TestClient(app)
|
||||
|
||||
mock_socket.clear_requests()
|
||||
fd = mock_socket.add_request('GET', '/baz')
|
||||
self._add_shutdown(app)
|
||||
app.run()
|
||||
self.assertTrue(fd.response.startswith(b'HTTP/1.0 200 OK\r\n'))
|
||||
self.assertIn(b'X-One: 1\r\n', fd.response)
|
||||
self.assertIn(b'Set-Cookie: foo=bar\r\n', fd.response)
|
||||
self.assertIn(b'Content-Length: 3\r\n', fd.response)
|
||||
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
|
||||
self.assertTrue(fd.response.endswith(b'\r\n\r\nbaz'))
|
||||
res = self._run(client.get('/bar'))
|
||||
self.assertEqual(res.status_code, 202)
|
||||
self.assertEqual(res.headers['Content-Type'], 'text/plain')
|
||||
self.assertEqual(res.headers['Set-Cookie'], ['foo=bar'])
|
||||
self.assertEqual(res.headers['X-One'], '1')
|
||||
self.assertEqual(res.headers['X-Two'], '2')
|
||||
self.assertEqual(res.text, 'bar')
|
||||
self.assertEqual(client.cookies['foo'], 'bar')
|
||||
|
||||
res = self._run(client.get('/baz'))
|
||||
self.assertEqual(res.status_code, 200)
|
||||
self.assertEqual(res.headers['Content-Type'], 'text/plain')
|
||||
self.assertEqual(res.headers['Set-Cookie'], ['foo=bar'])
|
||||
self.assertEqual(res.headers['X-One'], '1')
|
||||
self.assertFalse('X-Two' in res.headers)
|
||||
self.assertEqual(res.headers['Content-Length'], '3')
|
||||
self.assertEqual(res.text, 'baz')
|
||||
|
||||
def test_400(self):
|
||||
self._mock()
|
||||
|
||||
app = Microdot()
|
||||
|
||||
mock_socket.clear_requests()
|
||||
@@ -150,7 +312,11 @@ class TestMicrodotAsync(unittest.TestCase):
|
||||
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
|
||||
self.assertTrue(fd.response.endswith(b'\r\n\r\nBad request'))
|
||||
|
||||
self._unmock()
|
||||
|
||||
def test_400_handler(self):
|
||||
self._mock()
|
||||
|
||||
app = Microdot()
|
||||
|
||||
@app.errorhandler(400)
|
||||
@@ -167,6 +333,8 @@ class TestMicrodotAsync(unittest.TestCase):
|
||||
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
|
||||
self.assertTrue(fd.response.endswith(b'\r\n\r\n400'))
|
||||
|
||||
self._unmock()
|
||||
|
||||
def test_404(self):
|
||||
app = Microdot()
|
||||
|
||||
@@ -174,14 +342,11 @@ class TestMicrodotAsync(unittest.TestCase):
|
||||
def index(req):
|
||||
return 'foo'
|
||||
|
||||
mock_socket.clear_requests()
|
||||
fd = mock_socket.add_request('GET', '/foo')
|
||||
self._add_shutdown(app)
|
||||
app.run()
|
||||
self.assertTrue(fd.response.startswith(b'HTTP/1.0 404 N/A\r\n'))
|
||||
self.assertIn(b'Content-Length: 9\r\n', fd.response)
|
||||
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
|
||||
self.assertTrue(fd.response.endswith(b'\r\n\r\nNot found'))
|
||||
client = TestClient(app)
|
||||
res = self._run(client.post('/foo'))
|
||||
self.assertEqual(res.status_code, 404)
|
||||
self.assertEqual(res.headers['Content-Type'], 'text/plain')
|
||||
self.assertEqual(res.text, 'Not found')
|
||||
|
||||
def test_404_handler(self):
|
||||
app = Microdot()
|
||||
@@ -194,14 +359,11 @@ class TestMicrodotAsync(unittest.TestCase):
|
||||
async def handle_404(req):
|
||||
return '404'
|
||||
|
||||
mock_socket.clear_requests()
|
||||
fd = mock_socket.add_request('GET', '/foo')
|
||||
self._add_shutdown(app)
|
||||
app.run()
|
||||
self.assertTrue(fd.response.startswith(b'HTTP/1.0 200 OK\r\n'))
|
||||
self.assertIn(b'Content-Length: 3\r\n', fd.response)
|
||||
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
|
||||
self.assertTrue(fd.response.endswith(b'\r\n\r\n404'))
|
||||
client = TestClient(app)
|
||||
res = self._run(client.post('/foo'))
|
||||
self.assertEqual(res.status_code, 200)
|
||||
self.assertEqual(res.headers['Content-Type'], 'text/plain')
|
||||
self.assertEqual(res.text, '404')
|
||||
|
||||
def test_405(self):
|
||||
app = Microdot()
|
||||
@@ -210,14 +372,11 @@ class TestMicrodotAsync(unittest.TestCase):
|
||||
def index(req):
|
||||
return 'foo'
|
||||
|
||||
mock_socket.clear_requests()
|
||||
fd = mock_socket.add_request('POST', '/foo')
|
||||
self._add_shutdown(app)
|
||||
app.run()
|
||||
self.assertTrue(fd.response.startswith(b'HTTP/1.0 405 N/A\r\n'))
|
||||
self.assertIn(b'Content-Length: 9\r\n', fd.response)
|
||||
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
|
||||
self.assertTrue(fd.response.endswith(b'\r\n\r\nNot found'))
|
||||
client = TestClient(app)
|
||||
res = self._run(client.post('/foo'))
|
||||
self.assertEqual(res.status_code, 405)
|
||||
self.assertEqual(res.headers['Content-Type'], 'text/plain')
|
||||
self.assertEqual(res.text, 'Not found')
|
||||
|
||||
def test_405_handler(self):
|
||||
app = Microdot()
|
||||
@@ -228,16 +387,13 @@ class TestMicrodotAsync(unittest.TestCase):
|
||||
|
||||
@app.errorhandler(405)
|
||||
async def handle_405(req):
|
||||
return '405'
|
||||
return '405', 405
|
||||
|
||||
mock_socket.clear_requests()
|
||||
fd = mock_socket.add_request('POST', '/foo')
|
||||
self._add_shutdown(app)
|
||||
app.run()
|
||||
self.assertTrue(fd.response.startswith(b'HTTP/1.0 200 OK\r\n'))
|
||||
self.assertIn(b'Content-Length: 3\r\n', fd.response)
|
||||
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
|
||||
self.assertTrue(fd.response.endswith(b'\r\n\r\n405'))
|
||||
client = TestClient(app)
|
||||
res = self._run(client.patch('/foo'))
|
||||
self.assertEqual(res.status_code, 405)
|
||||
self.assertEqual(res.headers['Content-Type'], 'text/plain')
|
||||
self.assertEqual(res.text, '405')
|
||||
|
||||
def test_413(self):
|
||||
app = Microdot()
|
||||
@@ -246,14 +402,11 @@ class TestMicrodotAsync(unittest.TestCase):
|
||||
def index(req):
|
||||
return 'foo'
|
||||
|
||||
mock_socket.clear_requests()
|
||||
fd = mock_socket.add_request('GET', '/foo', body='x' * 17000)
|
||||
self._add_shutdown(app)
|
||||
app.run()
|
||||
self.assertTrue(fd.response.startswith(b'HTTP/1.0 413 N/A\r\n'))
|
||||
self.assertIn(b'Content-Length: 17\r\n', fd.response)
|
||||
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
|
||||
self.assertTrue(fd.response.endswith(b'\r\n\r\nPayload too large'))
|
||||
client = TestClient(app)
|
||||
res = self._run(client.post('/foo', body='x' * 17000))
|
||||
self.assertEqual(res.status_code, 413)
|
||||
self.assertEqual(res.headers['Content-Type'], 'text/plain')
|
||||
self.assertEqual(res.text, 'Payload too large')
|
||||
|
||||
def test_413_handler(self):
|
||||
app = Microdot()
|
||||
@@ -266,14 +419,11 @@ class TestMicrodotAsync(unittest.TestCase):
|
||||
async def handle_413(req):
|
||||
return '413', 400
|
||||
|
||||
mock_socket.clear_requests()
|
||||
fd = mock_socket.add_request('GET', '/foo', body='x' * 17000)
|
||||
self._add_shutdown(app)
|
||||
app.run()
|
||||
self.assertTrue(fd.response.startswith(b'HTTP/1.0 400 N/A\r\n'))
|
||||
self.assertIn(b'Content-Length: 3\r\n', fd.response)
|
||||
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
|
||||
self.assertTrue(fd.response.endswith(b'\r\n\r\n413'))
|
||||
client = TestClient(app)
|
||||
res = self._run(client.post('/foo', body='x' * 17000))
|
||||
self.assertEqual(res.status_code, 400)
|
||||
self.assertEqual(res.headers['Content-Type'], 'text/plain')
|
||||
self.assertEqual(res.text, '413')
|
||||
|
||||
def test_500(self):
|
||||
app = Microdot()
|
||||
@@ -282,14 +432,11 @@ class TestMicrodotAsync(unittest.TestCase):
|
||||
def index(req):
|
||||
return 1 / 0
|
||||
|
||||
mock_socket.clear_requests()
|
||||
fd = mock_socket.add_request('GET', '/')
|
||||
self._add_shutdown(app)
|
||||
app.run()
|
||||
self.assertTrue(fd.response.startswith(b'HTTP/1.0 500 N/A\r\n'))
|
||||
self.assertIn(b'Content-Length: 21\r\n', fd.response)
|
||||
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
|
||||
self.assertTrue(fd.response.endswith(b'\r\n\r\nInternal server error'))
|
||||
client = TestClient(app)
|
||||
res = self._run(client.get('/'))
|
||||
self.assertEqual(res.status_code, 500)
|
||||
self.assertEqual(res.headers['Content-Type'], 'text/plain')
|
||||
self.assertEqual(res.text, 'Internal server error')
|
||||
|
||||
def test_500_handler(self):
|
||||
app = Microdot()
|
||||
@@ -302,14 +449,11 @@ class TestMicrodotAsync(unittest.TestCase):
|
||||
def handle_500(req):
|
||||
return '501', 501
|
||||
|
||||
mock_socket.clear_requests()
|
||||
fd = mock_socket.add_request('GET', '/')
|
||||
self._add_shutdown(app)
|
||||
app.run()
|
||||
self.assertTrue(fd.response.startswith(b'HTTP/1.0 501 N/A\r\n'))
|
||||
self.assertIn(b'Content-Length: 3\r\n', fd.response)
|
||||
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
|
||||
self.assertTrue(fd.response.endswith(b'\r\n\r\n501'))
|
||||
client = TestClient(app)
|
||||
res = self._run(client.get('/'))
|
||||
self.assertEqual(res.status_code, 501)
|
||||
self.assertEqual(res.headers['Content-Type'], 'text/plain')
|
||||
self.assertEqual(res.text, '501')
|
||||
|
||||
def test_exception_handler(self):
|
||||
app = Microdot()
|
||||
@@ -322,14 +466,50 @@ class TestMicrodotAsync(unittest.TestCase):
|
||||
async def handle_div_zero(req, exc):
|
||||
return '501', 501
|
||||
|
||||
mock_socket.clear_requests()
|
||||
fd = mock_socket.add_request('GET', '/')
|
||||
self._add_shutdown(app)
|
||||
app.run()
|
||||
self.assertTrue(fd.response.startswith(b'HTTP/1.0 501 N/A\r\n'))
|
||||
self.assertIn(b'Content-Length: 3\r\n', fd.response)
|
||||
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
|
||||
self.assertTrue(fd.response.endswith(b'\r\n\r\n501'))
|
||||
client = TestClient(app)
|
||||
res = self._run(client.get('/'))
|
||||
self.assertEqual(res.status_code, 501)
|
||||
self.assertEqual(res.headers['Content-Type'], 'text/plain')
|
||||
self.assertEqual(res.text, '501')
|
||||
|
||||
def test_json_response(self):
|
||||
app = Microdot()
|
||||
|
||||
@app.route('/dict')
|
||||
async def json_dict(req):
|
||||
return {'foo': 'bar'}
|
||||
|
||||
@app.route('/list')
|
||||
def json_list(req):
|
||||
return ['foo', 'bar']
|
||||
|
||||
client = TestClient(app)
|
||||
|
||||
res = self._run(client.get('/dict'))
|
||||
self.assertEqual(res.status_code, 200)
|
||||
self.assertEqual(res.headers['Content-Type'], 'application/json')
|
||||
self.assertEqual(res.json, {'foo': 'bar'})
|
||||
|
||||
res = self._run(client.get('/list'))
|
||||
self.assertEqual(res.status_code, 200)
|
||||
self.assertEqual(res.headers['Content-Type'], 'application/json')
|
||||
self.assertEqual(res.json, ['foo', 'bar'])
|
||||
|
||||
def test_binary_response(self):
|
||||
app = Microdot()
|
||||
|
||||
@app.route('/bin')
|
||||
def index(req):
|
||||
return b'\xff\xfe', {'Content-Type': 'application/octet-stream'}
|
||||
|
||||
client = TestClient(app)
|
||||
res = self._run(client.get('/bin'))
|
||||
self.assertEqual(res.status_code, 200)
|
||||
self.assertEqual(res.headers['Content-Type'],
|
||||
'application/octet-stream')
|
||||
self.assertEqual(res.text, None)
|
||||
self.assertEqual(res.json, None)
|
||||
self.assertEqual(res.body, b'\xff\xfe')
|
||||
|
||||
def test_streaming(self):
|
||||
app = Microdot()
|
||||
@@ -353,10 +533,8 @@ class TestMicrodotAsync(unittest.TestCase):
|
||||
|
||||
return stream()
|
||||
|
||||
mock_socket.clear_requests()
|
||||
fd = mock_socket.add_request('GET', '/')
|
||||
self._add_shutdown(app)
|
||||
app.run()
|
||||
self.assertTrue(fd.response.startswith(b'HTTP/1.0 200 OK\r\n'))
|
||||
self.assertIn(b'Content-Type: text/plain\r\n', fd.response)
|
||||
self.assertTrue(fd.response.endswith(b'\r\n\r\nfoobar'))
|
||||
client = TestClient(app)
|
||||
res = self._run(client.get('/'))
|
||||
self.assertEqual(res.status_code, 200)
|
||||
self.assertEqual(res.headers['Content-Type'], 'text/plain')
|
||||
self.assertEqual(res.text, 'foobar')
|
||||
|
||||
Reference in New Issue
Block a user