Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
nonlocal startup_complete
startup_complete = True
@app.after_serving
async def on_shutdown():
nonlocal shutdown_complete
shutdown_complete = True
@app.route(path)
async def hello():
return "hello world!"
assert not startup_complete
assert not shutdown_complete
handler = Mangum(app)
assert startup_complete
assert not shutdown_complete
response = handler(mock_http_event, {})
assert response == {
"statusCode": 200,
"isBase64Encoded": False,
"headers": {"content-length": "12", "content-type": "text/html; charset=utf-8"},
"body": "hello world!",
}
assert startup_complete
assert shutdown_complete
@app.websocket_route("/ws")
async def websocket_endpoint(websocket):
await websocket.accept()
await websocket.send_json({"url": str(websocket.url)})
await websocket.close()
handler = Mangum(app, enable_lifespan=False)
response = handler(mock_ws_connect_event, {})
assert response == {
"body": "OK",
"headers": {"content-type": "text/plain; charset=utf-8"},
"isBase64Encoded": False,
"statusCode": 200,
}
handler = Mangum(app, enable_lifespan=False)
with mock.patch(
"mangum.protocols.websockets.ASGIWebSocketCycle.send_data"
) as send_data:
send_data.return_value = None
response = handler(mock_ws_send_event, {})
assert response == {
"body": "OK",
"headers": {"content-type": "text/plain; charset=utf-8"},
"isBase64Encoded": False,
"statusCode": 200,
}
def test_http_cycle_state(mock_http_event) -> None:
async def app(scope, receive, send):
assert scope["type"] == "http"
await send({"type": "http.response.body", "body": b"Hello, world!"})
handler = Mangum(app, enable_lifespan=False)
response = handler(mock_http_event, {})
assert response == {
"body": "Internal Server Error",
"headers": {"content-type": "text/plain; charset=utf-8"},
"isBase64Encoded": False,
"statusCode": 500,
}
async def app(scope, receive, send):
assert scope["type"] == "http"
await send({"type": "http.response.start", "status": 200})
await send({"type": "http.response.start", "status": 200})
handler = Mangum(app, enable_lifespan=False)
"raw_path": None,
"root_path": "",
"scheme": "https",
"server": ("test.execute-api.us-west-2.amazonaws.com", 80),
"type": "http",
}
await send(
{
"type": "http.response.start",
"status": 200,
"headers": [[b"content-type", b"text/plain; charset=utf-8"]],
}
)
await send({"type": "http.response.body", "body": b"Hello, world!"})
handler = Mangum(app, enable_lifespan=False)
response = handler(mock_http_event, {})
assert response == {
"statusCode": 200,
"isBase64Encoded": False,
"headers": {"content-type": "text/plain; charset=utf-8"},
"body": "Hello, world!",
}
}
handler = Mangum(app, enable_lifespan=False)
with mock.patch(
"mangum.protocols.websockets.ASGIWebSocketCycle.send_data"
) as send_data:
send_data.return_value = None
response = handler(mock_ws_send_event, {})
assert response == {
"body": "OK",
"headers": {"content-type": "text/plain; charset=utf-8"},
"isBase64Encoded": False,
"statusCode": 200,
}
handler = Mangum(app, enable_lifespan=False)
response = handler(mock_ws_disconnect_event, {})
assert response == {
"body": "OK",
"headers": {"content-type": "text/plain; charset=utf-8"},
"isBase64Encoded": False,
"statusCode": 200,
}
ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
)
async def app(scope, receive, send):
await send({"type": "websocket.send", "text": "Hello world!"})
handler = Mangum(app, enable_lifespan=False)
response = handler(mock_ws_connect_event, {})
assert response == {
"body": "OK",
"headers": {"content-type": "text/plain; charset=utf-8"},
"isBase64Encoded": False,
"statusCode": 200,
}
handler = Mangum(app, enable_lifespan=False)
with pytest.raises(RuntimeError):
with mock.patch(
"mangum.protocols.websockets.ASGIWebSocketCycle.send_data"
) as send_data:
send_data.return_value = None
handler(mock_ws_send_event, {})
def test_http_exception(mock_http_event) -> None:
async def app(scope, receive, send):
await send({"type": "http.response.start", "status": 200})
raise Exception()
await send({"type": "http.response.body", "body": b"1", "more_body": True})
handler = Mangum(app, enable_lifespan=False)
response = handler(mock_http_event, {})
assert response == {
"body": "Internal Server Error",
"headers": {"content-type": "text/plain; charset=utf-8"},
"isBase64Encoded": False,
"statusCode": 500,
}
nonlocal startup_complete
startup_complete = True
@app.on_event("shutdown")
async def on_shutdown():
nonlocal shutdown_complete
shutdown_complete = True
@app.route(path)
def homepage(request):
return PlainTextResponse("Hello, world!")
assert not startup_complete
assert not shutdown_complete
handler = Mangum(app)
mock_http_event["body"] = None
assert startup_complete
assert not shutdown_complete
response = handler(mock_http_event, {})
assert response == {
"statusCode": 200,
"isBase64Encoded": False,
"headers": {
"content-length": "13",
"content-type": "text/plain; charset=utf-8",
},
"body": "Hello, world!",
}
# @app.route("/hello")
# def homepage(request):
# return PlainTextResponse("hello world!")
# -- Quart -- #
# app = Quart(__name__)
# @app.route("/hello")
# async def hello():
# return "hello world!"
handler = Mangum(app, debug=True)
def transpose(content: str = Body(...), degree: int = Body(...)):
parser = Song(content, degree=degree)
return {"content": parser.text, "html_content": parser.html}
@app.post("/download")
def download(request: Request, content: str = Form(...)) -> StreamingResponse:
filename = f"guitarlette-{datetime.now().strftime('%Y-%m-%d-%H:%M:%S')}.txt"
return StreamingResponse(
BytesIO(content.encode()),
headers={"Content-Disposition": f"attachment; filename={filename}"},
media_type="text/plain",
)
handler = Mangum(app, lifespan="off")