test_depend_err_middleware() — fastapi Function Reference

Architecture documentation for the test_depend_err_middleware() function in test_ws_router.py from the fastapi codebase.

Function python FastAPI Routing calls 1

Entity Profile

FastAPI→ Routing→ test_depend_err_middleware() — fastapi Function Reference

Dependency Diagram

graph TD
  c70dd714_d3ca_bdb8_903b_f7f34bdd402a["test_depend_err_middleware()"]
  b2e4c2fb_8042_93dc_9d79_2ae1a12773d6["test_ws_router.py"]
  c70dd714_d3ca_bdb8_903b_f7f34bdd402a -->|defined in| b2e4c2fb_8042_93dc_9d79_2ae1a12773d6
  afb5e1a2_9bfc_75af_3be4_6639f67befbd["make_app()"]
  c70dd714_d3ca_bdb8_903b_f7f34bdd402a -->|calls| afb5e1a2_9bfc_75af_3be4_6639f67befbd
  style c70dd714_d3ca_bdb8_903b_f7f34bdd402a fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

tests/test_ws_router.py lines 236–254

def test_depend_err_middleware():
    """
    Verify that it is possible to write custom WebSocket middleware to catch errors
    """

    @websocket_middleware
    async def errorhandler(websocket: WebSocket, call_next):
        try:
            return await call_next()
        except Exception as e:
            await websocket.close(code=status.WS_1006_ABNORMAL_CLOSURE, reason=repr(e))

    myapp = make_app(middleware=[Middleware(errorhandler)])
    client = TestClient(myapp)
    with pytest.raises(WebSocketDisconnect) as e:
        with client.websocket_connect("/depends-err/"):
            pass  # pragma: no cover
    assert e.value.code == status.WS_1006_ABNORMAL_CLOSURE
    assert "NotImplementedError" in e.value.reason