Test coverage for serverless middleware

.coveragerc

@@ -5,3 +5,4 @@ omit =
     */templates/*
     */commands.py
     mangum/cli/__init__.py
+    tests/*

mangum/asgi/middleware.py

@@ -8,7 +8,7 @@ def __call__(self, *args, **kwargs):
             response = self.asgi(*args, **kwargs)
         except Exception as exc:
             if self.debug:
-                return self._debug(exc)
+                return self._debug(str(exc))
             raise exc
         else:
             return response

tests/test_asgi.py

@@ -1,5 +1,6 @@
 import pytest
 from mangum.asgi.protocol import ASGICycle
+from mangum.asgi.middleware import ServerlessMiddleware
 from starlette.responses import PlainTextResponse
 
 
@@ -11,25 +12,29 @@ def on_response_body(self, body: str) -> None:
         self.response["body"] = body
 
 
-def mock_asgi_handler(app, event: dict) -> dict:
-    scope = {
-        "type": "http",
-        "server": None,
-        "client": None,
-        "scheme": "https",
-        "root_path": "",
-        "query_string": "",
-        "headers": [],
-        "http_version": "1.1",
-        "method": "GET",
-        "path": "/",
-    }
-    body = event.get("body", b"")
-    handler = MockASGICycle(scope, body=body)
-    return handler(app)
-
-
-def test_asgi_handler() -> None:
+class MockServerlessMiddleware(ServerlessMiddleware):
+    def asgi(self, event: dict) -> dict:
+        scope = {
+            "type": "http",
+            "server": None,
+            "client": None,
+            "scheme": "https",
+            "root_path": "",
+            "query_string": "",
+            "headers": [],
+            "http_version": "1.1",
+            "method": "GET",
+            "path": "/",
+        }
+        body = event.get("body", b"")
+        handler = MockASGICycle(scope, body=body)
+        return handler(self.app)
+
+    def _debug(self, content: str, status_code: int = 500) -> dict:
+        return {"body": content, "status_code": status_code}
+
+
+def test_serverless_middleware() -> None:
     def app(scope):
         async def asgi(receive, send):
             res = PlainTextResponse("Hello, world!")
@@ -38,7 +43,7 @@ async def asgi(receive, send):
         return asgi
 
     mock_request = {}
-    response = mock_asgi_handler(app, mock_request)
+    response = MockServerlessMiddleware(app)(mock_request)
 
     assert response == {"status": 200, "body": b"Hello, world!"}
 
@@ -51,7 +56,7 @@ async def asgi(receive, send):
         return asgi
 
     with pytest.raises(RuntimeError):
-        mock_asgi_handler(app, {})
+        MockServerlessMiddleware(app)({})
 
     def app(scope):
         async def asgi(receive, send):
@@ -61,4 +66,31 @@ async def asgi(receive, send):
         return asgi
 
     with pytest.raises(RuntimeError):
-        mock_asgi_handler(app, {})
+        MockServerlessMiddleware(app)({})
+
+
+def test_serverless_middleware_not_implemented() -> None:
+    def app(scope):
+        async def asgi(receive, send):
+            res = PlainTextResponse("Hello, world!")
+            await res(receive, send)
+
+        return asgi
+
+    mock_request = {}
+    with pytest.raises(NotImplementedError):
+        ServerlessMiddleware(app)(mock_request)
+
+
+def test_serverless_middleware_debug() -> None:
+    def app(scope):
+        async def asgi(receive, send):
+            res = PlainTextResponse("Hello, world!")
+            raise Exception("There was an error!")
+            await res(receive, send)
+
+        return asgi
+
+    mock_request = {}
+    response = MockServerlessMiddleware(app, debug=True)(mock_request)
+    assert response == {"body": "There was an error!", "status_code": 500}

tests/test_aws_middleware.py

@@ -76,6 +76,26 @@ async def asgi(receive, send):
     }
 
 
+def test_aws_debug(mock_data) -> None:
+    def app(scope):
+        async def asgi(receive, send):
+            response = PlainTextResponse("Hello, world!")
+            raise Exception("Error!")
+            await response(receive, send)
+
+        return asgi
+
+    mock_event = mock_data.get_aws_event()
+    handler = AWSLambdaMiddleware(app, debug=True)
+    response = handler(mock_event, {})
+    assert response == {
+        "statusCode": 500,
+        "isBase64Encoded": False,
+        "headers": {},
+        "body": "Error!",
+    }
+
+
 def test_starlette_aws_response(mock_data) -> None:
 
     mock_event = mock_data.get_aws_event()