klen / asgi-tools Goto Github PK
View Code? Open in Web Editor NEWTools to build ASGI apps
License: MIT License
Tools to build ASGI apps
License: MIT License
ASGITestClient build_scope is missing "subprotocols".
https://asgi.readthedocs.io/en/latest/specs/www.html#websocket-connection-scope
...
subprotocols (Iterable[Unicode string]) โ Subprotocols the client advertised. Optional; if missing defaults to empty list.
Also, test client websocket example in the documentation is incorrect (it's missing websocket-context manager). In my case that would be something like this. For example code, the headers are not that important.
async def test_app(app):
client = ASGITestClient(app)
headers = {"Sec-WebSocket-Protocol": "ocpp1.6, ocpp2.0.1"}
async with client.websocket(path="/123", headers=headers) as ws:
await ws.send("ping")
msg = await ws.receive()
assert msg == "pong"
Anyway, nice library.
ASGITestClient doesn't send websocket close code, which is mandatory value in ASGI specification. Currently only "type" is send.
https://asgi.readthedocs.io/en/latest/specs/www.html#disconnect-receive-event-ws
async def disconnect(self):
await self.send({'type': 'websocket.disconnect'})
self.state = self.STATES.disconnected
WebSocket error codes:
https://datatracker.ietf.org/doc/html/rfc6455#section-7.4
Still found one more issue which this time is related to asyncio task not fully finished.
Example code to reproduce the problem, note the artifical asyncio.sleep, which is simulated to trigger the problem condition. Note that I had to comment out pytest.raises as if it's defined the problem condition is not triggered.
import asyncio
import pytest
from asgi_tools import App, ResponseWebSocket, ASGIConnectionClosed
from asgi_tools.tests import ASGITestClient
@pytest.fixture
def app():
app = App(debug=True)
@app.route('/')
async def index(request):
return 'OK'
return app
@pytest.fixture(scope="session")
def Client():
return ASGITestClient
@pytest.mark.asyncio
async def test_websocket(app, Client):
@app.route('/websocket')
async def websocket(request):
assert request.subprotocols == ['ship', 'done']
async with ResponseWebSocket(request) as ws:
msg = await ws.receive()
assert msg == 'ping'
await ws.send('pong')
await asyncio.sleep(1) # simulate executing some logic after sending response
async with Client(app).websocket(
'/websocket', headers={'sec-websocket-protocol': 'ship,done'}) as ws:
await ws.send('ping')
msg = await ws.receive()
assert msg == 'pong'
# with pytest.raises(ASGIConnectionClosed):
# await ws.receive()
Task was destroyed but it is pending!
task: <Task pending name='Task-2' coro=<App.call() done, defined at ...asgi_tools/app.py:196> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x1097c9070>()]>>
One possible fix for this is to wait in aio_spawn until task truly has finished. There might be alternative (and more proper) ways to fix this as well.
#...asgi_tools/_compat.py
@asynccontextmanager
async def aio_spawn(fn: t.Callable[..., t.Awaitable], *args, **kwargs):
"""Spawn a given coroutine."""
if trio and current_async_library() == 'trio':
async with trio.open_nursery() as tasks:
tasks.start_soon(fn, *args, **kwargs)
yield tasks
elif curio and current_async_library() == 'curio':
task = await curio.spawn(fn, *args, **kwargs)
yield task
await task.join()
else:
task = create_task(fn(*args, **kwargs))
yield task
while not task.done(): # Wait until task is truly finished
await asyncio.sleep(0.02)
There are wheels published for 3.8-3.11, but not yet 3.12.
A declarative, efficient, and flexible JavaScript library for building user interfaces.
๐ Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. ๐๐๐
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google โค๏ธ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.