self.app = app
def __call__(self, scope):
+ if scope["type"] != "http":
+ return self.app(scope)
return _DebugResponder(self.app, scope)
class _DebugResponder:
def __init__(self, app, scope):
+ self.app = app
self.scope = scope
- self.asgi_instance = app(scope)
self.response_started = False
async def __call__(self, receive, send):
self.raw_send = send
try:
- await self.asgi_instance(receive, self.send)
+ asgi = self.app(self.scope)
+ await asgi(receive, self.send)
except:
if self.response_started:
raise
client = TestClient(DebugMiddleware(app))
with pytest.raises(RuntimeError):
response = client.get("/")
+
+
+def test_debug_error_during_scope():
+ def app(scope):
+ raise RuntimeError("Something went wrong")
+
+ app = DebugMiddleware(app)
+ client = TestClient(DebugMiddleware(app))
+ response = client.get("/", headers={"Accept": "text/html, */*"})
+ assert response.status_code == 500
+ assert response.headers["content-type"].startswith("text/html")
+ assert "RuntimeError" in response.text
+
+
+def test_debug_not_http():
+ def app(scope):
+ raise RuntimeError("Something went wrong")
+
+ app = DebugMiddleware(app)
+
+ with pytest.raises(RuntimeError):
+ app({"type": "websocket"})