51 lines
1.2 KiB
Python
51 lines
1.2 KiB
Python
import asyncio
|
|
|
|
import pytest
|
|
from uvicorn.config import Config
|
|
from uvicorn.main import Server
|
|
|
|
|
|
async def app(scope, receive, send):
|
|
assert scope["type"] == "http"
|
|
if scope["path"] == "/slow_response":
|
|
await slow_response(scope, receive, send)
|
|
else:
|
|
await hello_world(scope, receive, send)
|
|
|
|
|
|
async def hello_world(scope, receive, send):
|
|
await send(
|
|
{
|
|
"type": "http.response.start",
|
|
"status": 200,
|
|
"headers": [[b"content-type", b"text/plain"]],
|
|
}
|
|
)
|
|
await send({"type": "http.response.body", "body": b"Hello, world!"})
|
|
|
|
|
|
async def slow_response(scope, receive, send):
|
|
await asyncio.sleep(0.01)
|
|
await send(
|
|
{
|
|
"type": "http.response.start",
|
|
"status": 200,
|
|
"headers": [[b"content-type", b"text/plain"]],
|
|
}
|
|
)
|
|
await send({"type": "http.response.body", "body": b"Hello, world!"})
|
|
|
|
|
|
@pytest.fixture
|
|
async def server():
|
|
config = Config(app=app, lifespan="off")
|
|
server = Server(config=config)
|
|
task = asyncio.ensure_future(server.serve())
|
|
try:
|
|
while not server.started:
|
|
await asyncio.sleep(0.0001)
|
|
yield server
|
|
finally:
|
|
server.should_exit = True
|
|
await task
|