TonBERRY/microdot
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v2 (#186)

Browse Source
This commit is contained in:
Miguel Grinberg
2023-12-22 20:26:07 +00:00
committed by GitHub
parent 7a329d98a8
commit 20ea305fe7
114 changed files with 3868 additions and 6410 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
tests/__init__.py
View File

@@ -1,14 +1,11 @@
from .test_multidict import TestMultiDict
from .test_request import TestRequest
from .test_response import TestResponse
from .test_url_pattern import TestURLPattern
from .test_microdot import TestMicrodot
from .test_microdot_websocket import TestMicrodotWebSocket
from .test_request_asyncio import TestRequestAsync
from .test_response_asyncio import TestResponseAsync
from .test_microdot_asyncio import TestMicrodotAsync
from .test_microdot_asyncio_websocket import TestMicrodotAsyncWebSocket
from .test_utemplate import TestUTemplate
from .test_session import TestSession
from tests.test_microdot import * # noqa: F401, F403
from tests.test_multidict import * # noqa: F401, F403
from tests.test_request import * # noqa: F401, F403
from tests.test_response import * # noqa: F401, F403
from tests.test_urlencode import * # noqa: F401, F403
from tests.test_url_pattern import * # noqa: F401, F403
from tests.test_websocket import * # noqa: F401, F403
from tests.test_sse import * # noqa: F401, F403
from tests.test_cors import * # noqa: F401, F403
from tests.test_utemplate import * # noqa: F401, F403
from tests.test_session import * # noqa: F401, F403

39
tests/mock_asyncio.py
View File

@@ -1,39 +0,0 @@
try:
import uasyncio as asyncio
except ImportError:
import asyncio
from tests import mock_socket
def get_event_loop():
return asyncio.get_event_loop()
async def start_server(cb, host, port):
class MockServer:
def __init__(self):
self.closed = False
async def run(self):
s = mock_socket.socket()
while not self.closed:
fd, addr = s.accept()
fd = mock_socket.FakeStreamAsync(fd)
await cb(fd, fd)
def close(self):
self.closed = True
async def wait_closed(self):
while not self.closed:
await asyncio.sleep(0.01)
server = MockServer()
asyncio.get_event_loop().create_task(server.run())
return server
def run(coro):
loop = asyncio.get_event_loop()
return loop.run_until_complete(coro)

36
tests/test_microdot_asgi.py → tests/test_asgi.py
View File

@@ -1,23 +1,23 @@
import unittest
import asyncio
import sys
import unittest
from unittest import mock
try:
import asyncio
except ImportError:
pass
try:
from unittest import mock
except ImportError:
mock = None
from microdot_asgi import Microdot, Response
from tests import mock_asyncio
from microdot.asgi import Microdot, Response
@unittest.skipIf(sys.implementation.name == 'micropython',
'not supported under MicroPython')
class TestMicrodotASGI(unittest.TestCase):
class TestASGI(unittest.TestCase):
@classmethod
def setUpClass(cls):
if hasattr(asyncio, 'set_event_loop'):
asyncio.set_event_loop(asyncio.new_event_loop())
cls.loop = asyncio.get_event_loop()
def _run(self, coro):
return self.loop.run_until_complete(coro)
def test_asgi_request_with_query_string(self):
app = Microdot()
@@ -98,7 +98,7 @@ class TestMicrodotASGI(unittest.TestCase):
original_buffer_size = Response.send_file_buffer_size
Response.send_file_buffer_size = 2
mock_asyncio.run(app(scope, receive, send))
self._run(app(scope, receive, send))
Response.send_file_buffer_size = original_buffer_size
@@ -143,7 +143,7 @@ class TestMicrodotASGI(unittest.TestCase):
async def send(packet):
pass
mock_asyncio.run(app(scope, receive, send))
self._run(app(scope, receive, send))
def test_shutdown(self):
app = Microdot()
@@ -166,7 +166,7 @@ class TestMicrodotASGI(unittest.TestCase):
async def send(packet):
pass
with mock.patch('microdot_asgi.os.kill') as kill:
mock_asyncio.run(app(scope, receive, send))
with mock.patch('microdot.asgi.os.kill') as kill:
self._run(app(scope, receive, send))
kill.assert_called()

51
tests/test_cors.py
View File

@@ -1,10 +1,18 @@
import asyncio
import unittest
from microdot import Microdot
from microdot_test_client import TestClient
from microdot_cors import CORS
from microdot.test_client import TestClient
from microdot.cors import CORS
class TestCORS(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.loop = asyncio.new_event_loop()
def _run(self, coro):
return self.loop.run_until_complete(coro)
def test_origin(self):
app = Microdot()
cors = CORS(allowed_origins=['https://example.com'],
@@ -16,13 +24,14 @@ class TestCORS(unittest.TestCase):
return 'foo'
client = TestClient(app)
res = client.get('/')
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 200)
self.assertFalse('Access-Control-Allow-Origin' in res.headers)
self.assertFalse('Access-Control-Allow-Credentials' in res.headers)
self.assertFalse('Vary' in res.headers)
res = client.get('/', headers={'Origin': 'https://example.com'})
res = self._run(client.get(
'/', headers={'Origin': 'https://example.com'}))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Access-Control-Allow-Origin'],
'https://example.com')
@@ -32,14 +41,15 @@ class TestCORS(unittest.TestCase):
cors.allow_credentials = False
res = client.get('/foo', headers={'Origin': 'https://example.com'})
res = self._run(client.get(
'/foo', headers={'Origin': 'https://example.com'}))
self.assertEqual(res.status_code, 404)
self.assertEqual(res.headers['Access-Control-Allow-Origin'],
'https://example.com')
self.assertFalse('Access-Control-Allow-Credentials' in res.headers)
self.assertEqual(res.headers['Vary'], 'Origin')
res = client.get('/', headers={'Origin': 'https://bad.com'})
res = self._run(client.get('/', headers={'Origin': 'https://bad.com'}))
self.assertEqual(res.status_code, 200)
self.assertFalse('Access-Control-Allow-Origin' in res.headers)
self.assertFalse('Access-Control-Allow-Credentials' in res.headers)
@@ -58,14 +68,15 @@ class TestCORS(unittest.TestCase):
return 'foo', {'Vary': 'X-Foo, X-Bar'}
client = TestClient(app)
res = client.get('/')
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Access-Control-Allow-Origin'], '*')
self.assertFalse('Vary' in res.headers)
self.assertEqual(res.headers['Access-Control-Expose-Headers'],
'X-Test, X-Test2')
res = client.get('/', headers={'Origin': 'https://example.com'})
res = self._run(client.get(
'/', headers={'Origin': 'https://example.com'}))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Access-Control-Allow-Origin'],
'https://example.com')
@@ -73,7 +84,8 @@ class TestCORS(unittest.TestCase):
self.assertEqual(res.headers['Access-Control-Expose-Headers'],
'X-Test, X-Test2')
res = client.get('/bad', headers={'Origin': 'https://example.com'})
res = self._run(client.get(
'/bad', headers={'Origin': 'https://example.com'}))
self.assertEqual(res.status_code, 404)
self.assertEqual(res.headers['Access-Control-Allow-Origin'],
'https://example.com')
@@ -81,7 +93,8 @@ class TestCORS(unittest.TestCase):
self.assertEqual(res.headers['Access-Control-Expose-Headers'],
'X-Test, X-Test2')
res = client.get('/foo', headers={'Origin': 'https://example.com'})
res = self._run(client.get(
'/foo', headers={'Origin': 'https://example.com'}))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Vary'], 'X-Foo, X-Bar, Origin')
@@ -94,10 +107,10 @@ class TestCORS(unittest.TestCase):
return 'foo'
client = TestClient(app)
res = client.request('OPTIONS', '/', headers={
res = self._run(client.request('OPTIONS', '/', headers={
'Origin': 'https://example.com',
'Access-Control-Request-Method': 'POST',
'Access-Control-Request-Headers': 'X-Test, X-Test2'})
'Access-Control-Request-Headers': 'X-Test, X-Test2'}))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Access-Control-Allow-Origin'],
'https://example.com')
@@ -106,8 +119,8 @@ class TestCORS(unittest.TestCase):
self.assertEqual(res.headers['Access-Control-Allow-Headers'],
'X-Test, X-Test2')
res = client.request('OPTIONS', '/', headers={
'Origin': 'https://example.com'})
res = self._run(client.request('OPTIONS', '/', headers={
'Origin': 'https://example.com'}))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Access-Control-Allow-Origin'],
'https://example.com')
@@ -125,10 +138,10 @@ class TestCORS(unittest.TestCase):
return 'foo'
client = TestClient(app)
res = client.request('OPTIONS', '/', headers={
res = self._run(client.request('OPTIONS', '/', headers={
'Origin': 'https://example.com',
'Access-Control-Request-Method': 'POST',
'Access-Control-Request-Headers': 'X-Test, X-Test2'})
'Access-Control-Request-Headers': 'X-Test, X-Test2'}))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Access-Control-Allow-Origin'],
'https://example.com')
@@ -136,9 +149,9 @@ class TestCORS(unittest.TestCase):
self.assertEqual(res.headers['Access-Control-Allow-Methods'], 'POST')
self.assertEqual(res.headers['Access-Control-Allow-Headers'], 'X-Test')
res = client.request('OPTIONS', '/', headers={
res = self._run(client.request('OPTIONS', '/', headers={
'Origin': 'https://example.com',
'Access-Control-Request-Method': 'GET'})
'Access-Control-Request-Method': 'GET'}))
self.assertEqual(res.status_code, 200)
self.assertFalse('Access-Control-Allow-Methods' in res.headers)
self.assertFalse('Access-Control-Allow-Headers' in res.headers)
@@ -152,7 +165,7 @@ class TestCORS(unittest.TestCase):
return 'foo'
client = TestClient(app)
res = client.get('/')
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 200)
self.assertFalse('Access-Control-Allow-Origin' in res.headers)
self.assertFalse('Vary' in res.headers)

104
tests/test_end2end.py Normal file
View File

@@ -0,0 +1,104 @@
import asyncio
import sys
import time
import unittest
from microdot import Microdot
class TestEnd2End(unittest.TestCase):
async def request(self, url, method='GET'):
while True:
reader, writer = await asyncio.open_connection('localhost', 5678)
try:
writer.write(f'{method} {url} HTTP/1.0\r\n\r\n'.encode())
break
except OSError:
# micropython's server sometimes needs a moment
writer.close()
await writer.wait_closed()
await asyncio.sleep(0.1)
await writer.drain()
response = await reader.read()
writer.close()
await writer.wait_closed()
return response.decode().splitlines()
def test_get(self):
app = Microdot()
@app.route('/')
def index(request):
return 'Hello, World!'
@app.route('/shutdown')
def shutdown(request):
app.shutdown()
return ''
async def run():
server = asyncio.create_task(app.start_server(host='127.0.0.1',
port=5678))
await asyncio.sleep(0.1)
response = await self.request('/')
self.assertEqual(response[0], 'HTTP/1.0 200 OK')
self.assertEqual(response[-1], 'Hello, World!')
await self.request('/shutdown')
await server
asyncio.run(run())
@unittest.skipIf(sys.implementation.name == 'micropython',
'not supported under MicroPython')
def test_concurrent_requests(self):
app = Microdot()
counter = 0
@app.route('/async1')
async def async1(request):
nonlocal counter
counter += 1
while counter < 4:
await asyncio.sleep(0.01)
return 'OK'
@app.route('/async2')
async def async2(request):
nonlocal counter
counter += 1
while counter < 4:
await asyncio.sleep(0.01)
return 'OK'
@app.route('/sync1')
def sync1(request):
nonlocal counter
counter += 1
while counter < 4:
time.sleep(0.01)
return 'OK'
@app.route('/sync2')
def sync2(request):
nonlocal counter
counter += 1
while counter < 4:
time.sleep(0.01)
return 'OK'
@app.route('/shutdown')
def shutdown(request):
app.shutdown()
return ''
async def run():
server = asyncio.create_task(app.start_server(port=5678))
await asyncio.sleep(0.1)
await asyncio.gather(self.request('/async1'),
self.request('/async2'),
self.request('/sync1'),
self.request('/sync2'))
await self.request('/shutdown')
await server
asyncio.run(run())

87
tests/test_jinja.py
View File

@@ -1,58 +1,81 @@
try:
import uasyncio as asyncio
except ImportError:
import asyncio
import asyncio
import sys
import unittest
from microdot import Microdot, Request
from microdot_asyncio import Microdot as MicrodotAsync, Request as RequestAsync
from microdot_jinja import render_template, init_templates
from tests.mock_socket import get_request_fd, get_async_request_fd
from microdot import Microdot
from microdot.jinja import Template, init_templates
from microdot.test_client import TestClient
init_templates('tests/templates')
def _run(coro):
return asyncio.get_event_loop().run_until_complete(coro)
@unittest.skipIf(sys.implementation.name == 'micropython',
'not supported under MicroPython')
class TestJinja(unittest.TestCase):
@classmethod
def setUpClass(cls):
if hasattr(asyncio, 'set_event_loop'):
asyncio.set_event_loop(asyncio.new_event_loop())
cls.loop = asyncio.get_event_loop()
def _run(self, coro):
return self.loop.run_until_complete(coro)
def test_render_template(self):
s = render_template('hello.jinja.txt', name='foo')
s = Template('hello.jinja.txt').render(name='foo')
self.assertEqual(s, 'Hello, foo!')
def test_render_template_in_app(self):
app = Microdot()
@app.route('/')
def index(req):
return render_template('hello.jinja.txt', name='foo')
async def index(req):
return Template('hello.jinja.txt').render(name='foo')
req = Request.create(app, get_request_fd('GET', '/'), 'addr')
res = app.dispatch_request(req)
client = TestClient(app)
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 200)
self.assertEqual(list(res.body_iter()), [b'Hello, foo!'])
self.assertEqual(res.body, b'Hello, foo!')
def test_render_template_in_app_async(self):
app = MicrodotAsync()
def test_generate_template_in_app(self):
app = Microdot()
@app.route('/')
async def index(req):
return render_template('hello.jinja.txt', name='foo')
return Template('hello.jinja.txt').generate(name='foo')
req = _run(RequestAsync.create(
app, get_async_request_fd('GET', '/'), 'writer', 'addr'))
res = _run(app.dispatch_request(req))
client = TestClient(app)
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.body, b'Hello, foo!')
async def get_result():
result = []
async for chunk in res.body_iter():
result.append(chunk)
return result
def test_render_async_template_in_app(self):
init_templates('tests/templates', enable_async=True)
result = _run(get_result())
self.assertEqual(result, [b'Hello, foo!'])
app = Microdot()
@app.route('/')
async def index(req):
return await Template('hello.jinja.txt').render_async(name='foo')
client = TestClient(app)
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.body, b'Hello, foo!')
init_templates('tests/templates')
def test_generate_async_template_in_app(self):
init_templates('tests/templates', enable_async=True)
app = Microdot()
@app.route('/')
async def index(req):
return Template('hello.jinja.txt').generate_async(name='foo')
client = TestClient(app)
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.body, b'Hello, foo!')
init_templates('tests/templates')

316
tests/test_microdot.py
View File

@@ -1,31 +1,18 @@
import sys
import asyncio
import unittest
from microdot import Microdot, Response, abort
from microdot_test_client import TestClient
from tests import mock_socket
from microdot.test_client import TestClient
class TestMicrodot(unittest.TestCase):
def _mock(self):
def mock_create_thread(f, *args, **kwargs):
f(*args, **kwargs)
@classmethod
def setUpClass(cls):
if hasattr(asyncio, 'set_event_loop'):
asyncio.set_event_loop(asyncio.new_event_loop())
cls.loop = asyncio.get_event_loop()
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(self):
sys.modules['microdot'].socket = self.original_socket
sys.modules['microdot'].create_thread = self.original_create_thread
def _add_shutdown(self, app):
@app.route('/shutdown')
def shutdown(req):
app.shutdown()
return ''
mock_socket.add_request('GET', '/shutdown')
def _run(self, coro):
return self.loop.run_until_complete(coro)
def test_get_request(self):
app = Microdot()
@@ -34,8 +21,13 @@ class TestMicrodot(unittest.TestCase):
def index(req):
return 'foo'
@app.route('/async')
async def index2(req):
return 'foo-async'
client = TestClient(app)
res = client.get('/')
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
@@ -44,6 +36,15 @@ class TestMicrodot(unittest.TestCase):
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; charset=UTF-8')
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()
@@ -55,78 +56,71 @@ class TestMicrodot(unittest.TestCase):
def index_post(req):
return Response('bar')
@app.route('/async', methods=['POST'])
async def index_post2(req):
return Response('bar-async')
client = TestClient(app)
res = client.post('/')
res = self._run(client.post('/'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
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; charset=UTF-8')
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_head_request(self):
self._mock()
app = Microdot()
@app.route('/foo')
def index(req):
return 'foo'
mock_socket.clear_requests()
fd = mock_socket.add_request('HEAD', '/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; charset=UTF-8\r\n',
fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\n'))
self._unmock()
client = TestClient(app)
res = self._run(client.request('HEAD', '/foo'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.headers['Content-Length'], '3')
self.assertIsNone(res.body)
self.assertIsNone(res.text)
self.assertIsNone(res.json)
def test_options_request(self):
app = Microdot()
@app.route('/', methods=['GET', 'DELETE'])
def index(req):
async def index(req):
return 'foo'
@app.post('/')
def index_post(req):
async def index_post(req):
return 'bar'
@app.route('/foo', methods=['POST', 'PUT'])
def foo(req):
async def foo(req):
return 'baz'
client = TestClient(app)
res = client.request('OPTIONS', '/')
res = self._run(client.request('OPTIONS', '/'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Allow'],
'GET, DELETE, POST, HEAD, OPTIONS')
res = client.request('OPTIONS', '/foo')
res = self._run(client.request('OPTIONS', '/foo'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Allow'], 'POST, PUT, OPTIONS')
def test_empty_request(self):
self._mock()
app = Microdot()
mock_socket.clear_requests()
fd = mock_socket.FakeStream(b'\n')
mock_socket._requests.append(fd)
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: 11\r\n', fd.response)
self.assertIn(b'Content-Type: text/plain; charset=UTF-8\r\n',
fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\nBad request'))
self._unmock()
def test_method_decorators(self):
app = Microdot()
@@ -135,7 +129,7 @@ class TestMicrodot(unittest.TestCase):
return 'GET'
@app.post('/post')
def post(req):
async def post(req):
return 'POST'
@app.put('/put')
@@ -143,7 +137,7 @@ class TestMicrodot(unittest.TestCase):
return 'PUT'
@app.patch('/patch')
def patch(req):
async def patch(req):
return 'PATCH'
@app.delete('/delete')
@@ -153,7 +147,8 @@ class TestMicrodot(unittest.TestCase):
client = TestClient(app)
methods = ['GET', 'POST', 'PUT', 'PATCH', 'DELETE']
for method in methods:
res = getattr(client, method.lower())('/' + method.lower())
res = self._run(getattr(
client, method.lower())('/' + method.lower()))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
@@ -167,7 +162,7 @@ class TestMicrodot(unittest.TestCase):
return req.headers.get('X-Foo')
client = TestClient(app)
res = client.get('/', headers={'X-Foo': 'bar'})
res = self._run(client.get('/', headers={'X-Foo': 'bar'}))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
@@ -178,15 +173,19 @@ class TestMicrodot(unittest.TestCase):
@app.route('/')
def index(req):
return req.cookies['one'] + req.cookies['two'] + \
req.cookies['three']
res = Response(
req.cookies['one'] + req.cookies['two'] + req.cookies['three'])
res.set_cookie('four', '4')
res.delete_cookie('two', path='/')
return res
client = TestClient(app, cookies={'one': '1', 'two': '2'})
res = client.get('/', headers={'Cookie': 'three=3'})
res = self._run(client.get('/', headers={'Cookie': 'three=3'}))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text, '123')
self.assertEqual(client.cookies, {'one': '1', 'four': '4'})
def test_binary_payload(self):
app = Microdot()
@@ -196,7 +195,7 @@ class TestMicrodot(unittest.TestCase):
return req.body
client = TestClient(app)
res = client.post('/', body=b'foo')
res = self._run(client.post('/', body=b'foo'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
@@ -216,13 +215,13 @@ class TestMicrodot(unittest.TestCase):
client = TestClient(app)
res = client.post('/dict', body={'foo': 'bar'})
res = self._run(client.post('/dict', body={'foo': 'bar'}))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text, 'bar')
res = client.post('/list', body=['foo', 'bar'])
res = self._run(client.post('/list', body=['foo', 'bar']))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
@@ -250,24 +249,24 @@ class TestMicrodot(unittest.TestCase):
client = TestClient(app)
res = client.get('/body')
res = self._run(client.get('/body'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text, 'one')
res = client.get('/body-status')
res = self._run(client.get('/body-status'))
self.assertEqual(res.status_code, 202)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text, 'two')
res = client.get('/body-headers')
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 = client.get('/body-status-headers')
res = self._run(client.get('/body-status-headers'))
self.assertEqual(res.status_code, 202)
self.assertEqual(res.headers['Content-Type'],
'text/html; charset=UTF-8')
@@ -280,7 +279,7 @@ class TestMicrodot(unittest.TestCase):
def before_request(req):
if req.path == '/bar':
@req.after_request
def after_request(req, res):
async def after_request(req, res):
res.headers['X-Two'] = '2'
return res
return 'bar', 202
@@ -291,7 +290,7 @@ class TestMicrodot(unittest.TestCase):
res.headers['X-One'] = '1'
@app.after_request
def after_request_two(req, res):
async def after_request_two(req, res):
res.set_cookie('foo', 'bar')
return res
@@ -305,7 +304,7 @@ class TestMicrodot(unittest.TestCase):
client = TestClient(app)
res = client.get('/bar')
res = self._run(client.get('/bar'))
self.assertEqual(res.status_code, 202)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
@@ -315,7 +314,7 @@ class TestMicrodot(unittest.TestCase):
self.assertEqual(res.text, 'bar')
self.assertEqual(client.cookies['foo'], 'bar')
res = client.get('/baz')
res = self._run(client.get('/baz'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
@@ -343,14 +342,14 @@ class TestMicrodot(unittest.TestCase):
client = TestClient(app)
res = client.get('/foo')
res = self._run(client.get('/foo'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertFalse('X-One' in res.headers)
self.assertFalse('Set-Cookie' in res.headers)
res = client.get('/bar')
res = self._run(client.get('/bar'))
self.assertEqual(res.status_code, 404)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
@@ -359,44 +358,22 @@ class TestMicrodot(unittest.TestCase):
self.assertEqual(client.cookies['foo'], 'bar')
def test_400(self):
self._mock()
app = Microdot()
mock_socket.clear_requests()
fd = mock_socket.FakeStream(b'\n')
mock_socket._requests.append(fd)
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: 11\r\n', fd.response)
self.assertIn(b'Content-Type: text/plain; charset=UTF-8\r\n',
fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\nBad request'))
self._unmock()
res = self._run(app.dispatch_request(None))
self.assertEqual(res.status_code, 400)
self.assertEqual(res.body, b'Bad request')
def test_400_handler(self):
self._mock()
app = Microdot()
@app.errorhandler(400)
def handle_400(req):
async def handle_400(req):
return '400'
mock_socket.clear_requests()
fd = mock_socket.FakeStream(b'\n')
mock_socket._requests.append(fd)
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; charset=UTF-8\r\n',
fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\n400'))
self._unmock()
res = self._run(app.dispatch_request(None))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.body, b'400')
def test_404(self):
app = Microdot()
@@ -406,7 +383,7 @@ class TestMicrodot(unittest.TestCase):
return 'foo'
client = TestClient(app)
res = client.post('/foo')
res = self._run(client.post('/foo'))
self.assertEqual(res.status_code, 404)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
@@ -420,11 +397,11 @@ class TestMicrodot(unittest.TestCase):
return 'foo'
@app.errorhandler(404)
def handle_404(req):
async def handle_404(req):
return '404'
client = TestClient(app)
res = client.post('/foo')
res = self._run(client.post('/foo'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
@@ -438,7 +415,7 @@ class TestMicrodot(unittest.TestCase):
return 'foo'
client = TestClient(app)
res = client.post('/foo')
res = self._run(client.post('/foo'))
self.assertEqual(res.status_code, 405)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
@@ -452,11 +429,11 @@ class TestMicrodot(unittest.TestCase):
return 'foo'
@app.errorhandler(405)
def handle_405(req):
async def handle_405(req):
return '405', 405
client = TestClient(app)
res = client.patch('/foo')
res = self._run(client.patch('/foo'))
self.assertEqual(res.status_code, 405)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
@@ -465,12 +442,12 @@ class TestMicrodot(unittest.TestCase):
def test_413(self):
app = Microdot()
@app.post('/')
@app.route('/')
def index(req):
return 'foo'
client = TestClient(app)
res = client.post('/foo', body='x' * 17000)
res = self._run(client.post('/foo', body='x' * 17000))
self.assertEqual(res.status_code, 413)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
@@ -484,11 +461,11 @@ class TestMicrodot(unittest.TestCase):
return 'foo'
@app.errorhandler(413)
def handle_413(req):
async def handle_413(req):
return '413', 400
client = TestClient(app)
res = client.post('/foo', body='x' * 17000)
res = self._run(client.post('/foo', body='x' * 17000))
self.assertEqual(res.status_code, 400)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
@@ -502,7 +479,7 @@ class TestMicrodot(unittest.TestCase):
return 1 / 0
client = TestClient(app)
res = client.get('/')
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 500)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
@@ -520,7 +497,7 @@ class TestMicrodot(unittest.TestCase):
return '501', 501
client = TestClient(app)
res = client.get('/')
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 501)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
@@ -534,11 +511,11 @@ class TestMicrodot(unittest.TestCase):
return 1 / 0
@app.errorhandler(ZeroDivisionError)
def handle_div_zero(req, exc):
async def handle_div_zero(req, exc):
return '501', 501
client = TestClient(app)
res = client.get('/')
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 501)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
@@ -553,11 +530,11 @@ class TestMicrodot(unittest.TestCase):
return foo[1]
@app.errorhandler(LookupError)
def handle_lookup_error(req, exc):
async def handle_lookup_error(req, exc):
return exc.__class__.__name__, 501
client = TestClient(app)
res = client.get('/')
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 501)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
@@ -572,15 +549,15 @@ class TestMicrodot(unittest.TestCase):
return foo[1]
@app.errorhandler(LookupError)
def handle_lookup_error(req, exc):
async def handle_lookup_error(req, exc):
return 'LookupError', 501
@app.errorhandler(IndexError)
def handle_index_error(req, exc):
async def handle_index_error(req, exc):
return 'IndexError', 501
client = TestClient(app)
res = client.get('/')
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 501)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
@@ -595,15 +572,15 @@ class TestMicrodot(unittest.TestCase):
return foo[1]
@app.errorhandler(Exception)
def handle_generic_exception(req, exc):
async def handle_generic_exception(req, exc):
return 'Exception', 501
@app.errorhandler(LookupError)
def handle_lookup_error(req, exc):
async def handle_lookup_error(req, exc):
return 'LookupError', 501
client = TestClient(app)
res = client.get('/')
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 501)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
@@ -618,11 +595,11 @@ class TestMicrodot(unittest.TestCase):
return foo[1]
@app.errorhandler(RuntimeError)
def handle_runtime_error(req, exc):
async def handle_runtime_error(req, exc):
return 'RuntimeError', 501
client = TestClient(app)
res = client.get('/')
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 500)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
@@ -637,7 +614,7 @@ class TestMicrodot(unittest.TestCase):
return 'foo'
client = TestClient(app)
res = client.get('/')
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 406)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
@@ -652,11 +629,11 @@ class TestMicrodot(unittest.TestCase):
return 'foo'
@app.errorhandler(406)
def handle_406(req):
def handle_500(req):
return '406', 406
client = TestClient(app)
res = client.get('/')
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 406)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
@@ -666,7 +643,7 @@ class TestMicrodot(unittest.TestCase):
app = Microdot()
@app.route('/dict')
def json_dict(req):
async def json_dict(req):
return {'foo': 'bar'}
@app.route('/list')
@@ -675,13 +652,13 @@ class TestMicrodot(unittest.TestCase):
client = TestClient(app)
res = client.get('/dict')
res = self._run(client.get('/dict'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'],
'application/json; charset=UTF-8')
self.assertEqual(res.json, {'foo': 'bar'})
res = client.get('/list')
res = self._run(client.get('/list'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'],
'application/json; charset=UTF-8')
@@ -695,7 +672,7 @@ class TestMicrodot(unittest.TestCase):
return b'\xff\xfe', {'Content-Type': 'application/octet-stream'}
client = TestClient(app)
res = client.get('/bin')
res = self._run(client.get('/bin'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'],
'application/octet-stream')
@@ -705,20 +682,38 @@ class TestMicrodot(unittest.TestCase):
def test_streaming(self):
app = Microdot()
done = False
@app.route('/')
def index(req):
def stream():
yield 'foo'
yield b'bar'
class stream():
def __init__(self):
self.i = 0
self.data = ['foo', b'bar']
def __aiter__(self):
return self
async def __anext__(self):
if self.i >= len(self.data):
raise StopAsyncIteration
data = self.data[self.i]
self.i += 1
return data
async def aclose(self):
nonlocal done
done = True
return stream()
client = TestClient(app)
res = client.get('/')
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text, 'foobar')
self.assertEqual(done, True)
def test_already_handled_response(self):
app = Microdot()
@@ -728,7 +723,7 @@ class TestMicrodot(unittest.TestCase):
return Response.already_handled
client = TestClient(app)
res = client.get('/')
res = self._run(client.get('/'))
self.assertEqual(res, None)
def test_mount(self):
@@ -759,39 +754,14 @@ class TestMicrodot(unittest.TestCase):
client = TestClient(app)
res = client.get('/app')
res = self._run(client.get('/app'))
self.assertEqual(res.status_code, 404)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text, '404:errorafter')
res = client.get('/sub/app')
res = self._run(client.get('/sub/app'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text, 'before:foo:after')
def test_ssl(self):
self._mock()
app = Microdot()
@app.route('/foo')
def foo(req):
return 'bar'
class FakeSSL:
def wrap_socket(self, sock, **kwargs):
return sock
mock_socket.clear_requests()
fd = mock_socket.add_request('GET', '/foo')
self._add_shutdown(app)
app.run(ssl=FakeSSL())
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; charset=UTF-8\r\n',
fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\nbar'))
self._unmock()

775
tests/test_microdot_asyncio.py
View File

@@ -1,775 +0,0 @@
try:
import uasyncio as asyncio
except ImportError:
import asyncio
import sys
import unittest
from microdot_asyncio import Microdot, Response, abort
from microdot_asyncio_test_client import TestClient
from tests import mock_asyncio, mock_socket
class TestMicrodotAsync(unittest.TestCase):
def setUp(self):
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 _unmock(self):
sys.modules['microdot_asyncio'].asyncio = self.original_asyncio
def _add_shutdown(self, app):
@app.route('/shutdown')
def shutdown(req):
app.shutdown()
return ''
mock_socket.add_request('GET', '/shutdown')
def test_get_request(self):
app = Microdot()
@app.route('/')
def index(req):
return 'foo'
@app.route('/async')
async def index2(req):
return 'foo-async'
client = TestClient(app)
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
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; charset=UTF-8')
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()
@app.route('/')
def index(req):
return 'foo'
@app.route('/', methods=['POST'])
def index_post(req):
return Response('bar')
@app.route('/async', methods=['POST'])
async def index_post2(req):
return Response('bar-async')
client = TestClient(app)
res = self._run(client.post('/'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
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; charset=UTF-8')
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_head_request(self):
app = Microdot()
@app.route('/foo')
def index(req):
return 'foo'
mock_socket.clear_requests()
fd = mock_socket.add_request('HEAD', '/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; charset=UTF-8\r\n',
fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\n'))
def test_options_request(self):
app = Microdot()
@app.route('/', methods=['GET', 'DELETE'])
async def index(req):
return 'foo'
@app.post('/')
async def index_post(req):
return 'bar'
@app.route('/foo', methods=['POST', 'PUT'])
async def foo(req):
return 'baz'
client = TestClient(app)
res = self._run(client.request('OPTIONS', '/'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Allow'],
'GET, DELETE, POST, HEAD, OPTIONS')
res = self._run(client.request('OPTIONS', '/foo'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Allow'], 'POST, PUT, OPTIONS')
def test_empty_request(self):
app = Microdot()
mock_socket.clear_requests()
fd = mock_socket.FakeStream(b'\n')
mock_socket._requests.append(fd)
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: 11\r\n', fd.response)
self.assertIn(b'Content-Type: text/plain; charset=UTF-8\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; charset=UTF-8')
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; charset=UTF-8')
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; charset=UTF-8')
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; charset=UTF-8')
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; charset=UTF-8')
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; charset=UTF-8')
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; charset=UTF-8'}
client = TestClient(app)
res = self._run(client.get('/body'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
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; charset=UTF-8')
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; charset=UTF-8')
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'
@app.after_request
def after_request_one(req, res):
res.headers['X-One'] = '1'
@app.after_request
async def after_request_two(req, res):
res.set_cookie('foo', 'bar')
return res
@app.route('/bar')
def bar(req):
return 'foo'
@app.route('/baz')
def baz(req):
return req.g.message
client = TestClient(app)
res = self._run(client.get('/bar'))
self.assertEqual(res.status_code, 202)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
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; charset=UTF-8')
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_after_error_request(self):
app = Microdot()
@app.after_error_request
def after_error_request_one(req, res):
res.headers['X-One'] = '1'
@app.after_error_request
def after_error_request_two(req, res):
res.set_cookie('foo', 'bar')
return res
@app.route('/foo')
def foo(req):
return 'foo'
client = TestClient(app)
res = self._run(client.get('/foo'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertFalse('X-One' in res.headers)
self.assertFalse('Set-Cookie' in res.headers)
res = self._run(client.get('/bar'))
self.assertEqual(res.status_code, 404)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.headers['Set-Cookie'], ['foo=bar'])
self.assertEqual(res.headers['X-One'], '1')
self.assertEqual(client.cookies['foo'], 'bar')
def test_400(self):
self._mock()
app = Microdot()
mock_socket.clear_requests()
fd = mock_socket.FakeStream(b'\n')
mock_socket._requests.append(fd)
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: 11\r\n', fd.response)
self.assertIn(b'Content-Type: text/plain; charset=UTF-8\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)
async def handle_404(req):
return '400'
mock_socket.clear_requests()
fd = mock_socket.FakeStream(b'\n')
mock_socket._requests.append(fd)
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; charset=UTF-8\r\n',
fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\n400'))
self._unmock()
def test_404(self):
app = Microdot()
@app.route('/')
def index(req):
return 'foo'
client = TestClient(app)
res = self._run(client.post('/foo'))
self.assertEqual(res.status_code, 404)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text, 'Not found')
def test_404_handler(self):
app = Microdot()
@app.route('/')
def index(req):
return 'foo'
@app.errorhandler(404)
async def handle_404(req):
return '404'
client = TestClient(app)
res = self._run(client.post('/foo'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text, '404')
def test_405(self):
app = Microdot()
@app.route('/foo')
def index(req):
return 'foo'
client = TestClient(app)
res = self._run(client.post('/foo'))
self.assertEqual(res.status_code, 405)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text, 'Not found')
def test_405_handler(self):
app = Microdot()
@app.route('/foo')
def index(req):
return 'foo'
@app.errorhandler(405)
async def handle_405(req):
return '405', 405
client = TestClient(app)
res = self._run(client.patch('/foo'))
self.assertEqual(res.status_code, 405)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text, '405')
def test_413(self):
app = Microdot()
@app.route('/')
def index(req):
return 'foo'
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; charset=UTF-8')
self.assertEqual(res.text, 'Payload too large')
def test_413_handler(self):
app = Microdot()
@app.route('/')
def index(req):
return 'foo'
@app.errorhandler(413)
async def handle_413(req):
return '413', 400
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; charset=UTF-8')
self.assertEqual(res.text, '413')
def test_500(self):
app = Microdot()
@app.route('/')
def index(req):
return 1 / 0
client = TestClient(app)
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 500)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text, 'Internal server error')
def test_500_handler(self):
app = Microdot()
@app.route('/')
def index(req):
return 1 / 0
@app.errorhandler(500)
def handle_500(req):
return '501', 501
client = TestClient(app)
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 501)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text, '501')
def test_exception_handler(self):
app = Microdot()
@app.route('/')
def index(req):
return 1 / 0
@app.errorhandler(ZeroDivisionError)
async def handle_div_zero(req, exc):
return '501', 501
client = TestClient(app)
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 501)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text, '501')
def test_exception_handler_parent(self):
app = Microdot()
@app.route('/')
def index(req):
foo = []
return foo[1]
@app.errorhandler(LookupError)
async def handle_lookup_error(req, exc):
return exc.__class__.__name__, 501
client = TestClient(app)
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 501)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text, 'IndexError')
def test_exception_handler_redundant_parent(self):
app = Microdot()
@app.route('/')
def index(req):
foo = []
return foo[1]
@app.errorhandler(LookupError)
async def handle_lookup_error(req, exc):
return 'LookupError', 501
@app.errorhandler(IndexError)
async def handle_index_error(req, exc):
return 'IndexError', 501
client = TestClient(app)
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 501)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text, 'IndexError')
def test_exception_handler_multiple_parents(self):
app = Microdot()
@app.route('/')
def index(req):
foo = []
return foo[1]
@app.errorhandler(Exception)
async def handle_generic_exception(req, exc):
return 'Exception', 501
@app.errorhandler(LookupError)
async def handle_lookup_error(req, exc):
return 'LookupError', 501
client = TestClient(app)
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 501)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text, 'LookupError')
def test_exception_handler_no_viable_parents(self):
app = Microdot()
@app.route('/')
def index(req):
foo = []
return foo[1]
@app.errorhandler(RuntimeError)
async def handle_runtime_error(req, exc):
return 'RuntimeError', 501
client = TestClient(app)
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 500)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text, 'Internal server error')
def test_abort(self):
app = Microdot()
@app.route('/')
def index(req):
abort(406, 'Not acceptable')
return 'foo'
client = TestClient(app)
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 406)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text, 'Not acceptable')
def test_abort_handler(self):
app = Microdot()
@app.route('/')
def index(req):
abort(406)
return 'foo'
@app.errorhandler(406)
def handle_500(req):
return '406', 406
client = TestClient(app)
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 406)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text, '406')
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; charset=UTF-8')
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; charset=UTF-8')
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()
@app.route('/')
def index(req):
class stream():
def __init__(self):
self.i = 0
self.data = ['foo', b'bar']
def __aiter__(self):
return self
async def __anext__(self):
if self.i >= len(self.data):
raise StopAsyncIteration
data = self.data[self.i]
self.i += 1
return data
return stream()
client = TestClient(app)
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text, 'foobar')
def test_already_handled_response(self):
app = Microdot()
@app.route('/')
def index(req):
return Response.already_handled
client = TestClient(app)
res = self._run(client.get('/'))
self.assertEqual(res, None)

71
tests/test_microdot_asyncio_websocket.py
View File

@@ -1,71 +0,0 @@
import sys
try:
import uasyncio as asyncio
except ImportError:
import asyncio
import unittest
from microdot_asyncio import Microdot
from microdot_asyncio_websocket import with_websocket
from microdot_asyncio_test_client import TestClient
class TestMicrodotAsyncWebSocket(unittest.TestCase):
def _run(self, coro):
loop = asyncio.get_event_loop()
return loop.run_until_complete(coro)
def test_websocket_echo(self):
app = Microdot()
@app.route('/echo')
@with_websocket
async def index(req, ws):
while True:
data = await ws.receive()
await ws.send(data)
results = []
def ws():
data = yield 'hello'
results.append(data)
data = yield b'bye'
results.append(data)
data = yield b'*' * 300
results.append(data)
data = yield b'+' * 65537
results.append(data)
client = TestClient(app)
res = self._run(client.websocket('/echo', ws))
self.assertIsNone(res)
self.assertEqual(results, ['hello', b'bye', b'*' * 300, b'+' * 65537])
@unittest.skipIf(sys.implementation.name == 'micropython',
'no support for async generators in MicroPython')
def test_websocket_echo_async_client(self):
app = Microdot()
@app.route('/echo')
@with_websocket
async def index(req, ws):
while True:
data = await ws.receive()
await ws.send(data)
results = []
async def ws():
data = yield 'hello'
results.append(data)
data = yield b'bye'
results.append(data)
data = yield b'*' * 300
results.append(data)
data = yield b'+' * 65537
results.append(data)
client = TestClient(app)
res = self._run(client.websocket('/echo', ws))
self.assertIsNone(res)
self.assertEqual(results, ['hello', b'bye', b'*' * 300, b'+' * 65537])

73
tests/test_microdot_websocket.py
View File

@@ -1,73 +0,0 @@
import unittest
from microdot import Microdot
from microdot_websocket import with_websocket, WebSocket
from microdot_test_client import TestClient
class TestMicrodotWebSocket(unittest.TestCase):
def test_websocket_echo(self):
app = Microdot()
@app.route('/echo')
@with_websocket
def index(req, ws):
while True:
data = ws.receive()
ws.send(data)
results = []
def ws():
data = yield 'hello'
results.append(data)
data = yield b'bye'
results.append(data)
data = yield b'*' * 300
results.append(data)
data = yield b'+' * 65537
results.append(data)
client = TestClient(app)
res = client.websocket('/echo', ws)
self.assertIsNone(res)
self.assertEqual(results, ['hello', b'bye', b'*' * 300, b'+' * 65537])
def test_bad_websocket_request(self):
app = Microdot()
@app.route('/echo')
@with_websocket
def index(req, ws):
return 'hello'
client = TestClient(app)
res = client.get('/echo')
self.assertEqual(res.status_code, 400)
res = client.get('/echo', headers={'Connection': 'Upgrade'})
self.assertEqual(res.status_code, 400)
res = client.get('/echo', headers={'Connection': 'foo'})
self.assertEqual(res.status_code, 400)
res = client.get('/echo', headers={'Upgrade': 'websocket'})
self.assertEqual(res.status_code, 400)
res = client.get('/echo', headers={'Upgrade': 'bar'})
self.assertEqual(res.status_code, 400)
res = client.get('/echo', headers={'Connection': 'Upgrade',
'Upgrade': 'websocket'})
self.assertEqual(res.status_code, 400)
res = client.get('/echo', headers={'Sec-WebSocket-Key': 'xxx'})
self.assertEqual(res.status_code, 400)
def test_process_websocket_frame(self):
ws = WebSocket(None)
ws.closed = True
self.assertEqual(ws._process_websocket_frame(WebSocket.TEXT, b'foo'),
(None, 'foo'))
self.assertEqual(ws._process_websocket_frame(WebSocket.BINARY, b'foo'),
(None, b'foo'))
self.assertRaises(OSError, ws._process_websocket_frame,
WebSocket.CLOSE, b'')
self.assertEqual(ws._process_websocket_frame(WebSocket.PING, b'foo'),
(WebSocket.PONG, b'foo'))
self.assertEqual(ws._process_websocket_frame(WebSocket.PONG, b'foo'),
(None, None))

2
tests/test_multidict.py
View File

@@ -1,5 +1,5 @@
import unittest
from microdot import MultiDict, NoCaseDict
from microdot.microdot import MultiDict, NoCaseDict
class TestMultiDict(unittest.TestCase):

95
tests/test_request.py
View File

@@ -1,12 +1,22 @@
import asyncio
import unittest
from microdot import Request, MultiDict
from tests.mock_socket import get_request_fd
from microdot.microdot import MultiDict, Request
from tests.mock_socket import get_async_request_fd
class TestRequest(unittest.TestCase):
@classmethod
def setUpClass(cls):
if hasattr(asyncio, 'set_event_loop'):
asyncio.set_event_loop(asyncio.new_event_loop())
cls.loop = asyncio.get_event_loop()
def _run(self, coro):
return self.loop.run_until_complete(coro)
def test_create_request(self):
fd = get_request_fd('GET', '/foo')
req = Request.create('app', fd, 'addr')
fd = get_async_request_fd('GET', '/foo')
req = self._run(Request.create('app', fd, 'writer', 'addr'))
self.assertEqual(req.app, 'app')
self.assertEqual(req.client_addr, 'addr')
self.assertEqual(req.method, 'GET')
@@ -23,11 +33,11 @@ class TestRequest(unittest.TestCase):
self.assertEqual(req.form, None)
def test_headers(self):
fd = get_request_fd('GET', '/foo', headers={
fd = get_async_request_fd('GET', '/foo', headers={
'Content-Type': 'application/json',
'Cookie': 'foo=bar;abc=def',
'Content-Length': '3'}, body='aaa')
req = Request.create('app', fd, 'addr')
req = self._run(Request.create('app', fd, 'writer', 'addr'))
self.assertEqual(req.headers, {
'Host': 'example.com:1234',
'Content-Type': 'application/json',
@@ -39,93 +49,68 @@ class TestRequest(unittest.TestCase):
self.assertEqual(req.body, b'aaa')
def test_args(self):
fd = get_request_fd('GET', '/?foo=bar&abc=def&foo&x=%2f%%')
req = Request.create('app', fd, 'addr')
fd = get_async_request_fd('GET', '/?foo=bar&abc=def&foo&x=%2f%%')
req = self._run(Request.create('app', fd, 'writer', 'addr'))
self.assertEqual(req.query_string, 'foo=bar&abc=def&foo&x=%2f%%')
md = MultiDict({'foo': 'bar', 'abc': 'def', 'x': '/%%'})
md['foo'] = ''
self.assertEqual(req.args, md)
def test_badly_formatted_args(self):
fd = get_request_fd('GET', '/?&foo=bar&abc=def&&&x=%2f%%')
req = Request.create('app', fd, 'addr')
self.assertEqual(req.query_string, '&foo=bar&abc=def&&&x=%2f%%')
self.assertEqual(req.args, MultiDict(
{'foo': 'bar', 'abc': 'def', 'x': '/%%'}))
def test_json(self):
fd = get_request_fd('GET', '/foo', headers={
fd = get_async_request_fd('GET', '/foo', headers={
'Content-Type': 'application/json'}, body='{"foo":"bar"}')
req = Request.create('app', fd, 'addr')
req = self._run(Request.create('app', fd, 'writer', 'addr'))
json = req.json
self.assertEqual(json, {'foo': 'bar'})
self.assertTrue(req.json is json)
fd = get_request_fd('GET', '/foo', headers={
fd = get_async_request_fd('GET', '/foo', headers={
'Content-Type': 'application/json'}, body='[1, "2"]')
req = Request.create('app', fd, 'addr')
req = self._run(Request.create('app', fd, 'writer', 'addr'))
self.assertEqual(req.json, [1, '2'])
fd = get_request_fd('GET', '/foo', headers={
fd = get_async_request_fd('GET', '/foo', headers={
'Content-Type': 'application/xml'}, body='[1, "2"]')
req = Request.create('app', fd, 'addr')
req = self._run(Request.create('app', fd, 'writer', 'addr'))
self.assertIsNone(req.json)
def test_form(self):
fd = get_request_fd('GET', '/foo', headers={
fd = get_async_request_fd('GET', '/foo', headers={
'Content-Type': 'application/x-www-form-urlencoded'},
body='foo=bar&abc=def&x=%2f%%')
req = Request.create('app', fd, 'addr')
req = self._run(Request.create('app', fd, 'writer', 'addr'))
form = req.form
self.assertEqual(form, MultiDict(
{'foo': 'bar', 'abc': 'def', 'x': '/%%'}))
self.assertTrue(req.form is form)
fd = get_request_fd('GET', '/foo', headers={
'Content-Type': 'application/x-www-form-urlencoded'},
body='')
req = Request.create('app', fd, 'addr')
form = req.form
self.assertEqual(form, MultiDict({}))
self.assertTrue(req.form is form)
fd = get_request_fd('GET', '/foo', headers={
fd = get_async_request_fd('GET', '/foo', headers={
'Content-Type': 'application/json'},
body='foo=bar&abc=def&x=%2f%%')
req = Request.create('app', fd, 'addr')
req = self._run(Request.create('app', fd, 'writer', 'addr'))
self.assertIsNone(req.form)
def test_large_line(self):
saved_max_readline = Request.max_readline
Request.max_readline = 16
fd = get_request_fd('GET', '/foo', headers={
fd = get_async_request_fd('GET', '/foo', headers={
'Content-Type': 'application/x-www-form-urlencoded'},
body='foo=bar&abc=def&x=y')
with self.assertRaises(ValueError):
Request.create('app', fd, 'addr')
self._run(Request.create('app', fd, 'writer', 'addr'))
Request.max_readline = saved_max_readline
def test_stream(self):
fd = get_request_fd('GET', '/foo', headers={
def test_body_and_stream(self):
fd = get_async_request_fd('GET', '/foo', headers={
'Content-Type': 'application/x-www-form-urlencoded',
'Content-Length': '19'},
body='foo=bar&abc=def&x=y')
req = Request.create('app', fd, 'addr')
self.assertEqual(req.stream.read(), b'foo=bar&abc=def&x=y')
with self.assertRaises(RuntimeError):
req.body
def test_body(self):
fd = get_request_fd('GET', '/foo', headers={
'Content-Type': 'application/x-www-form-urlencoded',
'Content-Length': '19'},
body='foo=bar&abc=def&x=y')
req = Request.create('app', fd, 'addr')
req = self._run(Request.create('app', fd, 'writer', 'addr'))
self.assertEqual(req.body, b'foo=bar&abc=def&x=y')
with self.assertRaises(RuntimeError):
req.stream
data = self._run(req.stream.read())
self.assertEqual(data, b'foo=bar&abc=def&x=y')
def test_large_payload(self):
saved_max_content_length = Request.max_content_length
@@ -133,12 +118,14 @@ class TestRequest(unittest.TestCase):
Request.max_content_length = 32
Request.max_body_length = 16
fd = get_request_fd('GET', '/foo', headers={
'Content-Type': 'application/x-www-form-urlencoded'},
fd = get_async_request_fd('GET', '/foo', headers={
'Content-Type': 'application/x-www-form-urlencoded',
'Content-Length': '19'},
body='foo=bar&abc=def&x=y')
req = Request.create('app', fd, 'addr')
req = self._run(Request.create('app', fd, 'writer', 'addr'))
self.assertEqual(req.body, b'')
self.assertEqual(req.stream.read(), b'foo=bar&abc=def&x=y')
data = self._run(req.stream.read())
self.assertEqual(data, b'foo=bar&abc=def&x=y')
Request.max_content_length = saved_max_content_length
Request.max_body_length = saved_max_body_length

131
tests/test_request_asyncio.py
View File

@@ -1,131 +0,0 @@
try:
import uasyncio as asyncio
except ImportError:
import asyncio
import unittest
from microdot import MultiDict
from microdot_asyncio import Request
from tests.mock_socket import get_async_request_fd
def _run(coro):
return asyncio.get_event_loop().run_until_complete(coro)
class TestRequestAsync(unittest.TestCase):
def test_create_request(self):
fd = get_async_request_fd('GET', '/foo')
req = _run(Request.create('app', fd, 'writer', 'addr'))
self.assertEqual(req.app, 'app')
self.assertEqual(req.client_addr, 'addr')
self.assertEqual(req.method, 'GET')
self.assertEqual(req.path, '/foo')
self.assertEqual(req.http_version, '1.0')
self.assertIsNone(req.query_string)
self.assertEqual(req.args, {})
self.assertEqual(req.headers, {'Host': 'example.com:1234'})
self.assertEqual(req.cookies, {})
self.assertEqual(req.content_length, 0)
self.assertEqual(req.content_type, None)
self.assertEqual(req.body, b'')
self.assertEqual(req.json, None)
self.assertEqual(req.form, None)
def test_headers(self):
fd = get_async_request_fd('GET', '/foo', headers={
'Content-Type': 'application/json',
'Cookie': 'foo=bar;abc=def',
'Content-Length': '3'}, body='aaa')
req = _run(Request.create('app', fd, 'writer', 'addr'))
self.assertEqual(req.headers, {
'Host': 'example.com:1234',
'Content-Type': 'application/json',
'Cookie': 'foo=bar;abc=def',
'Content-Length': '3'})
self.assertEqual(req.content_type, 'application/json')
self.assertEqual(req.cookies, {'foo': 'bar', 'abc': 'def'})
self.assertEqual(req.content_length, 3)
self.assertEqual(req.body, b'aaa')
def test_args(self):
fd = get_async_request_fd('GET', '/?foo=bar&abc=def&foo&x=%2f%%')
req = _run(Request.create('app', fd, 'writer', 'addr'))
self.assertEqual(req.query_string, 'foo=bar&abc=def&foo&x=%2f%%')
md = MultiDict({'foo': 'bar', 'abc': 'def', 'x': '/%%'})
md['foo'] = ''
self.assertEqual(req.args, md)
def test_json(self):
fd = get_async_request_fd('GET', '/foo', headers={
'Content-Type': 'application/json'}, body='{"foo":"bar"}')
req = _run(Request.create('app', fd, 'writer', 'addr'))
json = req.json
self.assertEqual(json, {'foo': 'bar'})
self.assertTrue(req.json is json)
fd = get_async_request_fd('GET', '/foo', headers={
'Content-Type': 'application/json'}, body='[1, "2"]')
req = _run(Request.create('app', fd, 'writer', 'addr'))
self.assertEqual(req.json, [1, '2'])
fd = get_async_request_fd('GET', '/foo', headers={
'Content-Type': 'application/xml'}, body='[1, "2"]')
req = _run(Request.create('app', fd, 'writer', 'addr'))
self.assertIsNone(req.json)
def test_form(self):
fd = get_async_request_fd('GET', '/foo', headers={
'Content-Type': 'application/x-www-form-urlencoded'},
body='foo=bar&abc=def&x=%2f%%')
req = _run(Request.create('app', fd, 'writer', 'addr'))
form = req.form
self.assertEqual(form, MultiDict(
{'foo': 'bar', 'abc': 'def', 'x': '/%%'}))
self.assertTrue(req.form is form)
fd = get_async_request_fd('GET', '/foo', headers={
'Content-Type': 'application/json'},
body='foo=bar&abc=def&x=%2f%%')
req = _run(Request.create('app', fd, 'writer', 'addr'))
self.assertIsNone(req.form)
def test_large_line(self):
saved_max_readline = Request.max_readline
Request.max_readline = 16
fd = get_async_request_fd('GET', '/foo', headers={
'Content-Type': 'application/x-www-form-urlencoded'},
body='foo=bar&abc=def&x=y')
with self.assertRaises(ValueError):
_run(Request.create('app', fd, 'writer', 'addr'))
Request.max_readline = saved_max_readline
def test_stream(self):
fd = get_async_request_fd('GET', '/foo', headers={
'Content-Type': 'application/x-www-form-urlencoded',
'Content-Length': '19'},
body='foo=bar&abc=def&x=y')
req = _run(Request.create('app', fd, 'writer', 'addr'))
self.assertEqual(req.body, b'foo=bar&abc=def&x=y')
data = _run(req.stream.read())
self.assertEqual(data, b'foo=bar&abc=def&x=y')
def test_large_payload(self):
saved_max_content_length = Request.max_content_length
saved_max_body_length = Request.max_body_length
Request.max_content_length = 32
Request.max_body_length = 16
fd = get_async_request_fd('GET', '/foo', headers={
'Content-Type': 'application/x-www-form-urlencoded',
'Content-Length': '19'},
body='foo=bar&abc=def&x=y')
req = _run(Request.create('app', fd, 'writer', 'addr'))
self.assertEqual(req.body, b'')
data = _run(req.stream.read())
self.assertEqual(data, b'foo=bar&abc=def&x=y')
Request.max_content_length = saved_max_content_length
Request.max_body_length = saved_max_body_length

201
tests/test_response.py
View File

@@ -1,67 +1,87 @@
import asyncio
from datetime import datetime
try:
import uio as io
except ImportError:
import io
import unittest
from microdot import Response
from tests.mock_socket import FakeStreamAsync
class TestResponse(unittest.TestCase):
@classmethod
def setUpClass(cls):
if hasattr(asyncio, 'set_event_loop'):
asyncio.set_event_loop(asyncio.new_event_loop())
cls.loop = asyncio.get_event_loop()
def _run(self, coro):
return self.loop.run_until_complete(coro)
def test_create_from_string(self):
res = Response('foo')
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers, {})
self.assertEqual(res.body, b'foo')
fd = io.BytesIO()
res.write(fd)
response = fd.getvalue()
self.assertIn(b'HTTP/1.0 200 OK\r\n', response)
self.assertIn(b'Content-Length: 3\r\n', response)
self.assertIn(b'Content-Type: text/plain; charset=UTF-8\r\n', response)
self.assertTrue(response.endswith(b'\r\n\r\nfoo'))
fd = FakeStreamAsync()
self._run(res.write(fd))
self.assertIn(b'HTTP/1.0 200 OK\r\n', fd.response)
self.assertIn(b'Content-Length: 3\r\n', fd.response)
self.assertIn(b'Content-Type: text/plain; charset=UTF-8\r\n',
fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\nfoo'))
def test_create_from_string_with_content_length(self):
res = Response('foo', headers={'Content-Length': '2'})
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers, {'Content-Length': '2'})
self.assertEqual(res.body, b'foo')
fd = io.BytesIO()
res.write(fd)
response = fd.getvalue()
self.assertIn(b'HTTP/1.0 200 OK\r\n', response)
self.assertIn(b'Content-Length: 2\r\n', response)
self.assertIn(b'Content-Type: text/plain; charset=UTF-8\r\n', response)
self.assertTrue(response.endswith(b'\r\n\r\nfoo'))
fd = FakeStreamAsync()
self._run(res.write(fd))
self.assertIn(b'HTTP/1.0 200 OK\r\n', fd.response)
self.assertIn(b'Content-Length: 2\r\n', fd.response)
self.assertIn(b'Content-Type: text/plain; charset=UTF-8\r\n',
fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\nfoo'))
def test_create_from_bytes(self):
res = Response(b'foo')
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers, {})
self.assertEqual(res.body, b'foo')
fd = io.BytesIO()
res.write(fd)
response = fd.getvalue()
self.assertIn(b'HTTP/1.0 200 OK\r\n', response)
self.assertIn(b'Content-Length: 3\r\n', response)
self.assertIn(b'Content-Type: text/plain; charset=UTF-8\r\n', response)
self.assertTrue(response.endswith(b'\r\n\r\nfoo'))
fd = FakeStreamAsync()
self._run(res.write(fd))
self.assertIn(b'HTTP/1.0 200 OK\r\n', fd.response)
self.assertIn(b'Content-Length: 3\r\n', fd.response)
self.assertIn(b'Content-Type: text/plain; charset=UTF-8\r\n',
fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\nfoo'))
def test_create_from_head(self):
res = Response(b'foo')
res.is_head = True
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers, {})
self.assertEqual(res.body, b'foo')
fd = FakeStreamAsync()
self._run(res.write(fd))
self.assertIn(b'HTTP/1.0 200 OK\r\n', fd.response)
self.assertIn(b'Content-Length: 3\r\n', fd.response)
self.assertIn(b'Content-Type: text/plain; charset=UTF-8\r\n',
fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\n'))
self.assertFalse(b'foo' in fd.response)
def test_create_empty(self):
res = Response(headers={'X-Foo': 'Bar'})
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers, {'X-Foo': 'Bar'})
self.assertEqual(res.body, b'')
fd = io.BytesIO()
res.write(fd)
response = fd.getvalue()
self.assertIn(b'HTTP/1.0 200 OK\r\n', response)
self.assertIn(b'X-Foo: Bar\r\n', response)
self.assertIn(b'Content-Length: 0\r\n', response)
self.assertIn(b'Content-Type: text/plain; charset=UTF-8\r\n', response)
self.assertTrue(response.endswith(b'\r\n\r\n'))
fd = FakeStreamAsync()
self._run(res.write(fd))
self.assertIn(b'HTTP/1.0 200 OK\r\n', fd.response)
self.assertIn(b'X-Foo: Bar\r\n', fd.response)
self.assertIn(b'Content-Length: 0\r\n', fd.response)
self.assertIn(b'Content-Type: text/plain; charset=UTF-8\r\n',
fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\n'))
def test_create_json(self):
res = Response({'foo': 'bar'})
@@ -69,40 +89,52 @@ class TestResponse(unittest.TestCase):
self.assertEqual(res.headers,
{'Content-Type': 'application/json; charset=UTF-8'})
self.assertEqual(res.body, b'{"foo": "bar"}')
fd = io.BytesIO()
res.write(fd)
response = fd.getvalue()
self.assertIn(b'HTTP/1.0 200 OK\r\n', response)
self.assertIn(b'Content-Length: 14\r\n', response)
fd = FakeStreamAsync()
self._run(res.write(fd))
self.assertIn(b'HTTP/1.0 200 OK\r\n', fd.response)
self.assertIn(b'Content-Length: 14\r\n', fd.response)
self.assertIn(b'Content-Type: application/json; charset=UTF-8\r\n',
response)
self.assertTrue(response.endswith(b'\r\n\r\n{"foo": "bar"}'))
fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\n{"foo": "bar"}'))
res = Response([1, '2'])
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers,
{'Content-Type': 'application/json; charset=UTF-8'})
self.assertEqual(res.body, b'[1, "2"]')
fd = io.BytesIO()
res.write(fd)
response = fd.getvalue()
self.assertIn(b'HTTP/1.0 200 OK\r\n', response)
self.assertIn(b'Content-Length: 8\r\n', response)
fd = FakeStreamAsync()
self._run(res.write(fd))
self.assertIn(b'HTTP/1.0 200 OK\r\n', fd.response)
self.assertIn(b'Content-Length: 8\r\n', fd.response)
self.assertIn(b'Content-Type: application/json; charset=UTF-8\r\n',
response)
self.assertTrue(response.endswith(b'\r\n\r\n[1, "2"]'))
fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\n[1, "2"]'))
def test_create_from_none(self):
res = Response(None)
self.assertEqual(res.status_code, 204)
self.assertEqual(res.body, b'')
fd = io.BytesIO()
res.write(fd)
response = fd.getvalue()
self.assertIn(b'HTTP/1.0 204 N/A\r\n', response)
self.assertIn(b'Content-Length: 0\r\n', response)
self.assertIn(b'Content-Type: text/plain; charset=UTF-8\r\n', response)
self.assertTrue(response.endswith(b'\r\n\r\n'))
fd = FakeStreamAsync()
self._run(res.write(fd))
self.assertIn(b'HTTP/1.0 204 N/A\r\n', fd.response)
self.assertIn(b'Content-Length: 0\r\n', fd.response)
self.assertIn(b'Content-Type: text/plain; charset=UTF-8\r\n',
fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\n'))
def test_create_from_iterator(self):
def gen():
yield 'foo'
yield 'bar'
res = Response(gen())
fd = FakeStreamAsync()
self._run(res.write(fd))
print(fd.response)
self.assertIn(b'HTTP/1.0 200 OK\r\n', fd.response)
self.assertIn(b'Content-Type: text/plain; charset=UTF-8\r\n',
fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\nfoobar'))
def test_create_from_other(self):
res = Response(123)
@@ -134,10 +166,9 @@ class TestResponse(unittest.TestCase):
self.assertEqual(res.headers, {})
self.assertEqual(res.reason, 'ALL GOOD!')
self.assertEqual(res.body, b'foo')
fd = io.BytesIO()
res.write(fd)
response = fd.getvalue()
self.assertIn(b'HTTP/1.0 200 ALL GOOD!\r\n', response)
fd = FakeStreamAsync()
self._run(res.write(fd))
self.assertIn(b'HTTP/1.0 200 ALL GOOD!\r\n', fd.response)
def test_create_with_status_and_reason(self):
res = Response('not found', 404, reason='NOT FOUND')
@@ -145,15 +176,14 @@ class TestResponse(unittest.TestCase):
self.assertEqual(res.headers, {})
self.assertEqual(res.reason, 'NOT FOUND')
self.assertEqual(res.body, b'not found')
fd = io.BytesIO()
res.write(fd)
response = fd.getvalue()
self.assertIn(b'HTTP/1.0 404 NOT FOUND\r\n', response)
fd = FakeStreamAsync()
self._run(res.write(fd))
self.assertIn(b'HTTP/1.0 404 NOT FOUND\r\n', fd.response)
def test_cookies(self):
res = Response('ok')
res.set_cookie('foo1', 'bar1')
res.set_cookie('foo2', 'bar2', path='/')
res.set_cookie('foo2', 'bar2', path='/', partitioned=True)
res.set_cookie('foo3', 'bar3', domain='example.com:1234')
res.set_cookie('foo4', 'bar4',
expires=datetime(2019, 11, 5, 2, 23, 54))
@@ -163,16 +193,18 @@ class TestResponse(unittest.TestCase):
res.set_cookie('foo7', 'bar7', path='/foo', domain='example.com:1234',
expires=datetime(2019, 11, 5, 2, 23, 54), max_age=123,
secure=True, http_only=True)
res.delete_cookie('foo8', http_only=True)
self.assertEqual(res.headers, {'Set-Cookie': [
'foo1=bar1',
'foo2=bar2; Path=/',
'foo2=bar2; Path=/; Partitioned',
'foo3=bar3; Domain=example.com:1234',
'foo4=bar4; Expires=Tue, 05 Nov 2019 02:23:54 GMT',
'foo5=bar5; Expires=Thu, 01 Jan 1970 00:00:00 GMT; Max-Age=123',
'foo6=bar6; Secure; HttpOnly',
'foo7=bar7; Path=/foo; Domain=example.com:1234; '
'Expires=Tue, 05 Nov 2019 02:23:54 GMT; Max-Age=123; Secure; '
'HttpOnly'
'HttpOnly',
'foo8=; Expires=Thu, 01 Jan 1970 00:00:01 GMT; HttpOnly',
]})
def test_redirect(self):
@@ -188,36 +220,14 @@ class TestResponse(unittest.TestCase):
Response.redirect('/foo\x0d\x0a\x0d\x0a<p>Foo</p>')
def test_send_file(self):
files = [
('test.txt', 'text/plain'),
('test.gif', 'image/gif'),
('test.jpg', 'image/jpeg'),
('test.png', 'image/png'),
('test.html', 'text/html'),
('test.css', 'text/css'),
('test.js', 'application/javascript'),
('test.json', 'application/json'),
('test.bin', 'application/octet-stream'),
]
for file, content_type in files:
res = Response.send_file('tests/files/' + file)
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'], content_type)
fd = io.BytesIO()
res.write(fd)
response = fd.getvalue()
self.assertEqual(response, (
b'HTTP/1.0 200 OK\r\nContent-Type: ' + content_type.encode()
+ b'\r\n\r\nfoo\n'))
res = Response.send_file('tests/files/test.txt',
content_type='text/html')
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'], 'text/html')
fd = io.BytesIO()
res.write(fd)
response = fd.getvalue()
fd = FakeStreamAsync()
self._run(res.write(fd))
self.assertEqual(
response,
fd.response,
b'HTTP/1.0 200 OK\r\nContent-Type: text/html\r\n\r\nfoo\n')
def test_send_file_small_buffer(self):
@@ -227,11 +237,10 @@ class TestResponse(unittest.TestCase):
content_type='text/html')
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'], 'text/html')
fd = io.BytesIO()
res.write(fd)
response = fd.getvalue()
fd = FakeStreamAsync()
self._run(res.write(fd))
self.assertEqual(
response,
fd.response,
b'HTTP/1.0 200 OK\r\nContent-Type: text/html\r\n\r\nfoo\n')
Response.send_file_buffer_size = original_buffer_size

139
tests/test_response_asyncio.py
View File

@@ -1,139 +0,0 @@
try:
import uasyncio as asyncio
except ImportError:
import asyncio
import unittest
from microdot_asyncio import Response
from tests.mock_socket import FakeStreamAsync
def _run(coro):
return asyncio.get_event_loop().run_until_complete(coro)
class TestResponseAsync(unittest.TestCase):
def test_create_from_string(self):
res = Response('foo')
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers, {})
self.assertEqual(res.body, b'foo')
fd = FakeStreamAsync()
_run(res.write(fd))
self.assertIn(b'HTTP/1.0 200 OK\r\n', fd.response)
self.assertIn(b'Content-Length: 3\r\n', fd.response)
self.assertIn(b'Content-Type: text/plain; charset=UTF-8\r\n',
fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\nfoo'))
def test_create_from_string_with_content_length(self):
res = Response('foo', headers={'Content-Length': '2'})
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers, {'Content-Length': '2'})
self.assertEqual(res.body, b'foo')
fd = FakeStreamAsync()
_run(res.write(fd))
self.assertIn(b'HTTP/1.0 200 OK\r\n', fd.response)
self.assertIn(b'Content-Length: 2\r\n', fd.response)
self.assertIn(b'Content-Type: text/plain; charset=UTF-8\r\n',
fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\nfoo'))
def test_create_from_bytes(self):
res = Response(b'foo')
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers, {})
self.assertEqual(res.body, b'foo')
fd = FakeStreamAsync()
_run(res.write(fd))
self.assertIn(b'HTTP/1.0 200 OK\r\n', fd.response)
self.assertIn(b'Content-Length: 3\r\n', fd.response)
self.assertIn(b'Content-Type: text/plain; charset=UTF-8\r\n',
fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\nfoo'))
def test_create_empty(self):
res = Response(headers={'X-Foo': 'Bar'})
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers, {'X-Foo': 'Bar'})
self.assertEqual(res.body, b'')
fd = FakeStreamAsync()
_run(res.write(fd))
self.assertIn(b'HTTP/1.0 200 OK\r\n', fd.response)
self.assertIn(b'X-Foo: Bar\r\n', fd.response)
self.assertIn(b'Content-Length: 0\r\n', fd.response)
self.assertIn(b'Content-Type: text/plain; charset=UTF-8\r\n',
fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\n'))
def test_create_json(self):
res = Response({'foo': 'bar'})
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers,
{'Content-Type': 'application/json; charset=UTF-8'})
self.assertEqual(res.body, b'{"foo": "bar"}')
fd = FakeStreamAsync()
_run(res.write(fd))
self.assertIn(b'HTTP/1.0 200 OK\r\n', fd.response)
self.assertIn(b'Content-Length: 14\r\n', fd.response)
self.assertIn(b'Content-Type: application/json; charset=UTF-8\r\n',
fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\n{"foo": "bar"}'))
res = Response([1, '2'])
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers,
{'Content-Type': 'application/json; charset=UTF-8'})
self.assertEqual(res.body, b'[1, "2"]')
fd = FakeStreamAsync()
_run(res.write(fd))
self.assertIn(b'HTTP/1.0 200 OK\r\n', fd.response)
self.assertIn(b'Content-Length: 8\r\n', fd.response)
self.assertIn(b'Content-Type: application/json; charset=UTF-8\r\n',
fd.response)
self.assertTrue(fd.response.endswith(b'\r\n\r\n[1, "2"]'))
def test_create_with_reason(self):
res = Response('foo', reason='ALL GOOD!')
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers, {})
self.assertEqual(res.reason, 'ALL GOOD!')
self.assertEqual(res.body, b'foo')
fd = FakeStreamAsync()
_run(res.write(fd))
self.assertIn(b'HTTP/1.0 200 ALL GOOD!\r\n', fd.response)
def test_create_with_status_and_reason(self):
res = Response('not found', 404, reason='NOT FOUND')
self.assertEqual(res.status_code, 404)
self.assertEqual(res.headers, {})
self.assertEqual(res.reason, 'NOT FOUND')
self.assertEqual(res.body, b'not found')
fd = FakeStreamAsync()
_run(res.write(fd))
self.assertIn(b'HTTP/1.0 404 NOT FOUND\r\n', fd.response)
def test_send_file(self):
res = Response.send_file('tests/files/test.txt',
content_type='text/html')
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'], 'text/html')
fd = FakeStreamAsync()
_run(res.write(fd))
self.assertEqual(
fd.response,
b'HTTP/1.0 200 OK\r\nContent-Type: text/html\r\n\r\nfoo\n')
def test_send_file_small_buffer(self):
original_buffer_size = Response.send_file_buffer_size
Response.send_file_buffer_size = 2
res = Response.send_file('tests/files/test.txt',
content_type='text/html')
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'], 'text/html')
fd = FakeStreamAsync()
_run(res.write(fd))
self.assertEqual(
fd.response,
b'HTTP/1.0 200 OK\r\nContent-Type: text/html\r\n\r\nfoo\n')
Response.send_file_buffer_size = original_buffer_size

103
tests/test_session.py
View File

@@ -1,79 +1,31 @@
try:
import uasyncio as asyncio
except ImportError:
import asyncio
import asyncio
import unittest
from microdot import Microdot
from microdot_asyncio import Microdot as MicrodotAsync
from microdot_session import set_session_secret_key, get_session, \
update_session, delete_session, with_session
from microdot_test_client import TestClient
from microdot_asyncio_test_client import TestClient as TestClientAsync
from microdot.session import Session, with_session
from microdot.test_client import TestClient
set_session_secret_key('top-secret!')
session_ext = Session(secret_key='top-secret!')
class TestSession(unittest.TestCase):
def test_session(self):
@classmethod
def setUpClass(cls):
if hasattr(asyncio, 'set_event_loop'):
asyncio.set_event_loop(asyncio.new_event_loop())
cls.loop = asyncio.get_event_loop()
def _run(self, coro):
return self.loop.run_until_complete(coro)
def test_session_async(self):
app = Microdot()
session_ext.initialize(app, secret_key='some-other-secret')
client = TestClient(app)
@app.get('/')
def index(req):
session = get_session(req)
session2 = get_session(req)
session2['foo'] = 'bar'
self.assertEqual(session['foo'], 'bar')
return str(session.get('name'))
@app.get('/with')
@with_session
def session_context_manager(req, session):
return str(session.get('name'))
@app.post('/set')
def set_session(req):
update_session(req, {'name': 'joe'})
return 'OK'
@app.post('/del')
def del_session(req):
delete_session(req)
return 'OK'
res = client.get('/')
self.assertEqual(res.text, 'None')
res = client.get('/with')
self.assertEqual(res.text, 'None')
res = client.post('/set')
self.assertEqual(res.text, 'OK')
res = client.get('/')
self.assertEqual(res.text, 'joe')
res = client.get('/with')
self.assertEqual(res.text, 'joe')
res = client.post('/del')
self.assertEqual(res.text, 'OK')
res = client.get('/')
self.assertEqual(res.text, 'None')
res = client.get('/with')
self.assertEqual(res.text, 'None')
def _run(self, coro):
loop = asyncio.get_event_loop()
return loop.run_until_complete(coro)
def test_session_async(self):
app = MicrodotAsync()
client = TestClientAsync(app)
@app.get('/')
async def index(req):
session = get_session(req)
session2 = get_session(req)
session = session_ext.get(req)
session2 = session_ext.get(req)
session2['foo'] = 'bar'
self.assertEqual(session['foo'], 'bar')
return str(session.get('name'))
@@ -84,13 +36,16 @@ class TestSession(unittest.TestCase):
return str(session.get('name'))
@app.post('/set')
async def set_session(req):
update_session(req, {'name': 'joe'})
@with_session
async def save_session(req, session):
session['name'] = 'joe'
session.save()
return 'OK'
@app.post('/del')
async def del_session(req):
delete_session(req)
@with_session
async def delete_session(req, session):
session.delete()
return 'OK'
res = self._run(client.get('/'))
@@ -115,17 +70,15 @@ class TestSession(unittest.TestCase):
self.assertEqual(res.text, 'None')
def test_session_no_secret_key(self):
set_session_secret_key(None)
app = Microdot()
session_ext = Session(app)
client = TestClient(app)
@app.get('/')
def index(req):
self.assertRaises(ValueError, get_session, req)
self.assertRaises(ValueError, update_session, req, {})
self.assertRaises(ValueError, session_ext.get, req)
self.assertRaises(ValueError, session_ext.update, req, {})
return ''
res = client.get('/')
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 200)
set_session_secret_key('top-secret!')

38
tests/test_sse.py Normal file
View File

@@ -0,0 +1,38 @@
import asyncio
import unittest
from microdot import Microdot
from microdot.sse import with_sse
from microdot.test_client import TestClient
class TestWebSocket(unittest.TestCase):
@classmethod
def setUpClass(cls):
if hasattr(asyncio, 'set_event_loop'):
asyncio.set_event_loop(asyncio.new_event_loop())
cls.loop = asyncio.get_event_loop()
def _run(self, coro):
return self.loop.run_until_complete(coro)
def test_sse(self):
app = Microdot()
@app.route('/sse')
@with_sse
async def handle_sse(request, sse):
await sse.send('foo')
await sse.send('bar', event='test')
await sse.send({'foo': 'bar'})
await sse.send([42, 'foo', 'bar'])
await sse.send(ValueError('foo'))
client = TestClient(app)
response = self._run(client.get('/sse'))
self.assertEqual(response.status_code, 200)
self.assertEqual(response.headers['Content-Type'], 'text/event-stream')
self.assertEqual(response.text, ('data: foo\n\n'
'event: test\ndata: bar\n\n'
'data: {"foo": "bar"}\n\n'
'data: [42, "foo", "bar"]\n\n'
'data: foo\n\n'))

2
tests/test_url_pattern.py
View File

@@ -1,5 +1,5 @@
import unittest
from microdot import URLPattern
from microdot.microdot import URLPattern
class TestURLPattern(unittest.TestCase):

2
tests/test_urlencode.py
View File

@@ -1,5 +1,5 @@
import unittest
from microdot import urlencode, urldecode_str, urldecode_bytes
from microdot.microdot import urlencode, urldecode_str, urldecode_bytes
class TestURLEncode(unittest.TestCase):

62
tests/test_utemplate.py
View File

@@ -1,55 +1,47 @@
try:
import uasyncio as asyncio
except ImportError:
import asyncio
import asyncio
import unittest
from microdot import Microdot, Request
from microdot_asyncio import Microdot as MicrodotAsync, Request as RequestAsync
from microdot_utemplate import render_template, init_templates
from tests.mock_socket import get_request_fd, get_async_request_fd
from microdot import Microdot
from microdot.test_client import TestClient
from microdot.utemplate import Template, init_templates
init_templates('tests/templates')
def _run(coro):
return asyncio.get_event_loop().run_until_complete(coro)
class TestUTemplate(unittest.TestCase):
@classmethod
def setUpClass(cls):
if hasattr(asyncio, 'set_event_loop'):
asyncio.set_event_loop(asyncio.new_event_loop())
cls.loop = asyncio.get_event_loop()
def _run(self, coro):
return self.loop.run_until_complete(coro)
def test_render_template(self):
s = list(render_template('hello.utemplate.txt', name='foo'))
self.assertEqual(s, ['Hello, ', 'foo', '!\n'])
s = Template('hello.utemplate.txt').render(name='foo')
self.assertEqual(s, 'Hello, foo!\n')
def test_render_template_in_app(self):
app = Microdot()
@app.route('/')
def index(req):
return render_template('hello.utemplate.txt', name='foo')
async def index(req):
return Template('hello.utemplate.txt').render(name='foo')
req = Request.create(app, get_request_fd('GET', '/'), 'addr')
res = app.dispatch_request(req)
client = TestClient(app)
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 200)
self.assertEqual(list(res.body_iter()), ['Hello, ', 'foo', '!\n'])
self.assertEqual(res.body, b'Hello, foo!\n')
def test_render_template_in_app_async(self):
app = MicrodotAsync()
def test_render_async_template_in_app(self):
app = Microdot()
@app.route('/')
async def index(req):
return render_template('hello.utemplate.txt', name='foo')
return await Template('hello.utemplate.txt').render_async(
name='foo')
req = _run(RequestAsync.create(
app, get_async_request_fd('GET', '/'), 'writer', 'addr'))
res = _run(app.dispatch_request(req))
client = TestClient(app)
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 200)
async def get_result():
result = []
async for chunk in res.body_iter():
result.append(chunk)
return result
result = _run(get_result())
self.assertEqual(result, ['Hello, ', 'foo', '!\n'])
self.assertEqual(res.body, b'Hello, foo!\n')

114
tests/test_websocket.py Normal file
View File

@@ -0,0 +1,114 @@
import asyncio
import sys
import unittest
from microdot import Microdot
from microdot.websocket import with_websocket, WebSocket
from microdot.test_client import TestClient
class TestWebSocket(unittest.TestCase):
@classmethod
def setUpClass(cls):
if hasattr(asyncio, 'set_event_loop'):
asyncio.set_event_loop(asyncio.new_event_loop())
cls.loop = asyncio.get_event_loop()
def _run(self, coro):
return self.loop.run_until_complete(coro)
def test_websocket_echo(self):
app = Microdot()
@app.route('/echo')
@with_websocket
async def index(req, ws):
while True:
data = await ws.receive()
await ws.send(data)
results = []
def ws():
data = yield 'hello'
results.append(data)
data = yield b'bye'
results.append(data)
data = yield b'*' * 300
results.append(data)
data = yield b'+' * 65537
results.append(data)
client = TestClient(app)
res = self._run(client.websocket('/echo', ws))
self.assertIsNone(res)
self.assertEqual(results, ['hello', b'bye', b'*' * 300, b'+' * 65537])
@unittest.skipIf(sys.implementation.name == 'micropython',
'no support for async generators in MicroPython')
def test_websocket_echo_async_client(self):
app = Microdot()
@app.route('/echo')
@with_websocket
async def index(req, ws):
while True:
data = await ws.receive()
await ws.send(data)
results = []
async def ws():
data = yield 'hello'
results.append(data)
data = yield b'bye'
results.append(data)
data = yield b'*' * 300
results.append(data)
data = yield b'+' * 65537
results.append(data)
client = TestClient(app)
res = self._run(client.websocket('/echo', ws))
self.assertIsNone(res)
self.assertEqual(results, ['hello', b'bye', b'*' * 300, b'+' * 65537])
def test_bad_websocket_request(self):
app = Microdot()
@app.route('/echo')
@with_websocket
def index(req, ws):
return 'hello'
client = TestClient(app)
res = self._run(client.get('/echo'))
self.assertEqual(res.status_code, 400)
res = self._run(client.get('/echo', headers={'Connection': 'Upgrade'}))
self.assertEqual(res.status_code, 400)
res = self._run(client.get('/echo', headers={'Connection': 'foo'}))
self.assertEqual(res.status_code, 400)
res = self._run(client.get('/echo', headers={'Upgrade': 'websocket'}))
self.assertEqual(res.status_code, 400)
res = self._run(client.get('/echo', headers={'Upgrade': 'bar'}))
self.assertEqual(res.status_code, 400)
res = self._run(client.get('/echo', headers={'Connection': 'Upgrade',
'Upgrade': 'websocket'}))
self.assertEqual(res.status_code, 400)
res = self._run(client.get(
'/echo', headers={'Sec-WebSocket-Key': 'xxx'}))
self.assertEqual(res.status_code, 400)
def test_process_websocket_frame(self):
ws = WebSocket(None)
ws.closed = True
self.assertEqual(ws._process_websocket_frame(WebSocket.TEXT, b'foo'),
(None, 'foo'))
self.assertEqual(ws._process_websocket_frame(WebSocket.BINARY, b'foo'),
(None, b'foo'))
self.assertRaises(OSError, ws._process_websocket_frame,
WebSocket.CLOSE, b'')
self.assertEqual(ws._process_websocket_frame(WebSocket.PING, b'foo'),
(WebSocket.PONG, b'foo'))
self.assertEqual(ws._process_websocket_frame(WebSocket.PONG, b'foo'),
(None, None))

62
tests/test_microdot_wsgi.py → tests/test_wsgi.py
View File

@@ -1,23 +1,15 @@
import unittest
import io
import sys
import unittest
from unittest import mock
try:
import uio as io
except ImportError:
import io
try:
from unittest import mock
except ImportError:
mock = None
from microdot_wsgi import Microdot
from microdot.wsgi import Microdot, Request
@unittest.skipIf(sys.implementation.name == 'micropython',
'not supported under MicroPython')
class TestMicrodotWSGI(unittest.TestCase):
def test_wsgi_request_with_query_string(self):
class TestWSGI(unittest.TestCase):
def test_request_with_query_string(self):
app = Microdot()
@app.post('/foo/bar')
@@ -29,6 +21,8 @@ class TestMicrodotWSGI(unittest.TestCase):
self.assertEqual(req.path, '/foo/bar')
self.assertEqual(req.args, {'baz': ['1']})
self.assertEqual(req.cookies, {'session': 'xyz'})
self.assertEqual(req.headers['Content-Length'], '4')
self.assertEqual(req.headers['Content-Type'], 'text/plain')
self.assertEqual(req.body, b'body')
return 'response'
@@ -43,7 +37,8 @@ class TestMicrodotWSGI(unittest.TestCase):
'QUERY_STRING': 'baz=1',
'HTTP_AUTHORIZATION': 'Bearer 123',
'HTTP_COOKIE': 'session=xyz',
'HTTP_CONTENT_LENGTH': '4',
'CONTENT_LENGTH': '4',
'CONTENT_TYPE': 'text/plain',
'REMOTE_ADDR': '1.2.3.4',
'REMOTE_PORT': '1234',
'REQUEST_METHOD': 'POST',
@@ -62,9 +57,9 @@ class TestMicrodotWSGI(unittest.TestCase):
self.assertIn(header, headers)
r = app(environ, start_response)
self.assertEqual(next(r), b'response')
self.assertEqual(b''.join(r), b'response')
def test_wsgi_request_without_query_string(self):
def test_request_without_query_string(self):
app = Microdot()
@app.route('/foo/bar')
@@ -88,6 +83,37 @@ class TestMicrodotWSGI(unittest.TestCase):
app(environ, start_response)
def test_request_with_stream(self):
saved_max_body_length = Request.max_body_length
Request.max_body_length = 2
app = Microdot()
@app.post('/foo/bar')
async def index(req):
self.assertEqual(req.body, b'')
self.assertEqual(await req.stream.read(), b'body')
return 'response'
environ = {
'SCRIPT_NAME': '/foo',
'PATH_INFO': '/bar',
'CONTENT_LENGTH': '4',
'CONTENT_TYPE': 'text/plain',
'REMOTE_ADDR': '1.2.3.4',
'REMOTE_PORT': '1234',
'REQUEST_METHOD': 'POST',
'SERVER_PROTOCOL': 'HTTP/1.1',
'wsgi.input': io.BytesIO(b'body'),
}
def start_response(status, headers):
pass
app(environ, start_response)
Request.max_body_length = saved_max_body_length
def test_shutdown(self):
app = Microdot()
@@ -107,7 +133,7 @@ class TestMicrodotWSGI(unittest.TestCase):
def start_response(status, headers):
pass
with mock.patch('microdot_wsgi.os.kill') as kill:
with mock.patch('microdot.wsgi.os.kill') as kill:
app(environ, start_response)
kill.assert_called()