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

Websocket standard and asyncio extensions (#55)

Browse Source
This commit is contained in:
Miguel Grinberg
2022-09-03 20:04:34 +01:00
committed by GitHub
parent ec0f9ba855
commit 2399c29c8a
27 changed files with 1077 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
tests/__init__.py
View File

@@ -3,11 +3,12 @@ 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

4
tests/test_jinja.py
View File

@@ -19,7 +19,7 @@ def _run(coro):
@unittest.skipIf(sys.implementation.name == 'micropython',
'not supported under MicroPython')
class TestUTemplate(unittest.TestCase):
class TestJinja(unittest.TestCase):
def test_render_template(self):
s = render_template('hello.jinja.txt', name='foo')
self.assertEqual(s, 'Hello, foo!')
@@ -44,7 +44,7 @@ class TestUTemplate(unittest.TestCase):
return render_template('hello.jinja.txt', name='foo')
req = _run(RequestAsync.create(
app, get_async_request_fd('GET', '/'), 'addr'))
app, get_async_request_fd('GET', '/'), 'writer', 'addr'))
res = _run(app.dispatch_request(req))
self.assertEqual(res.status_code, 200)

11
tests/test_microdot.py
View File

@@ -526,6 +526,17 @@ class TestMicrodot(unittest.TestCase):
self.assertEqual(res.headers['Content-Type'], 'text/plain')
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 = client.get('/')
self.assertEqual(res, None)
def test_mount(self):
subapp = Microdot()

11
tests/test_microdot_asyncio.py
View File

@@ -570,3 +570,14 @@ class TestMicrodotAsync(unittest.TestCase):
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'], 'text/plain')
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 Normal file
View File

@@ -0,0 +1,71 @@
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 Normal file
View File

@@ -0,0 +1,73 @@
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))

22
tests/test_request_asyncio.py
View File

@@ -16,7 +16,7 @@ def _run(coro):
class TestRequestAsync(unittest.TestCase):
def test_create_request(self):
fd = get_async_request_fd('GET', '/foo')
req = _run(Request.create('app', fd, 'addr'))
req = _run(Request.create('app', fd, 'writer', 'addr'))
self.assertEqual(req.app, 'app')
self.assertEqual(req.client_addr, 'addr')
self.assertEqual(req.method, 'GET')
@@ -37,7 +37,7 @@ class TestRequestAsync(unittest.TestCase):
'Content-Type': 'application/json',
'Cookie': 'foo=bar;abc=def',
'Content-Length': '3'}, body='aaa')
req = _run(Request.create('app', fd, 'addr'))
req = _run(Request.create('app', fd, 'writer', 'addr'))
self.assertEqual(req.headers, {
'Host': 'example.com:1234',
'Content-Type': 'application/json',
@@ -50,7 +50,7 @@ class TestRequestAsync(unittest.TestCase):
def test_args(self):
fd = get_async_request_fd('GET', '/?foo=bar&abc=def&x=%2f%%')
req = _run(Request.create('app', fd, 'addr'))
req = _run(Request.create('app', fd, 'writer', 'addr'))
self.assertEqual(req.query_string, 'foo=bar&abc=def&x=%2f%%')
self.assertEqual(req.args, MultiDict(
{'foo': 'bar', 'abc': 'def', 'x': '/%%'}))
@@ -58,26 +58,26 @@ class TestRequestAsync(unittest.TestCase):
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, 'addr'))
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, 'addr'))
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, 'addr'))
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, 'addr'))
req = _run(Request.create('app', fd, 'writer', 'addr'))
form = req.form
self.assertEqual(form, MultiDict(
{'foo': 'bar', 'abc': 'def', 'x': '/%%'}))
@@ -86,7 +86,7 @@ class TestRequestAsync(unittest.TestCase):
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, 'addr'))
req = _run(Request.create('app', fd, 'writer', 'addr'))
self.assertIsNone(req.form)
def test_large_line(self):
@@ -97,7 +97,7 @@ class TestRequestAsync(unittest.TestCase):
'Content-Type': 'application/x-www-form-urlencoded'},
body='foo=bar&abc=def&x=y')
with self.assertRaises(ValueError):
_run(Request.create('app', fd, 'addr'))
_run(Request.create('app', fd, 'writer', 'addr'))
Request.max_readline = saved_max_readline
@@ -106,7 +106,7 @@ class TestRequestAsync(unittest.TestCase):
'Content-Type': 'application/x-www-form-urlencoded',
'Content-Length': '19'},
body='foo=bar&abc=def&x=y')
req = _run(Request.create('app', fd, 'addr'))
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')
@@ -121,7 +121,7 @@ class TestRequestAsync(unittest.TestCase):
'Content-Type': 'application/x-www-form-urlencoded',
'Content-Length': '19'},
body='foo=bar&abc=def&x=y')
req = _run(Request.create('app', fd, 'addr'))
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')

2
tests/test_utemplate.py
View File

@@ -41,7 +41,7 @@ class TestUTemplate(unittest.TestCase):
return render_template('hello.utemplate.txt', name='foo')
req = _run(RequestAsync.create(
app, get_async_request_fd('GET', '/'), 'addr'))
app, get_async_request_fd('GET', '/'), 'writer', 'addr'))
res = _run(app.dispatch_request(req))
self.assertEqual(res.status_code, 200)