add test for middleware exception handling

Author: amirreza8002

tests/test_middlewares/middlewares.py

@@ -0,0 +1,135 @@
+from asgiref.sync import iscoroutinefunction, markcoroutinefunction
+
+from django.http import Http404, HttpResponse
+from django.template import engines
+from django.template.response import TemplateResponse
+from django.utils.decorators import (
+    async_only_middleware,
+    sync_and_async_middleware,
+)
+
+log = []
+
+
+class BaseMiddleware:
+    async_capable = True
+    sync_capable = False
+
+    def __init__(self, get_response):
+        self.get_response = get_response
+        if iscoroutinefunction(self.get_response):
+            markcoroutinefunction(self)
+
+    async def __call__(self, request):
+        return await self.get_response(request)
+
+
+class ProcessExceptionMiddleware(BaseMiddleware):
+    def process_exception(self, request, exception):
+        return HttpResponse("Exception caught")
+
+
+@async_only_middleware
+class AsyncProcessExceptionMiddleware(BaseMiddleware):
+    async def process_exception(self, request, exception):
+        return HttpResponse("Exception caught")
+
+
+class ProcessExceptionLogMiddleware(BaseMiddleware):
+    def process_exception(self, request, exception):
+        log.append("process-exception")
+
+
+class ProcessExceptionExcMiddleware(BaseMiddleware):
+    def process_exception(self, request, exception):
+        raise Exception("from process-exception")
+
+
+class ProcessViewMiddleware(BaseMiddleware):
+    def process_view(self, request, view_func, view_args, view_kwargs):
+        return HttpResponse("Processed view %s" % view_func.__name__)
+
+
+@async_only_middleware
+class AsyncProcessViewMiddleware(BaseMiddleware):
+    async def process_view(self, request, view_func, view_args, view_kwargs):
+        return HttpResponse("Processed view %s" % view_func.__name__)
+
+
+class ProcessViewNoneMiddleware(BaseMiddleware):
+    def process_view(self, request, view_func, view_args, view_kwargs):
+        log.append("processed view %s" % view_func.__name__)
+        return None
+
+
+class ProcessViewTemplateResponseMiddleware(BaseMiddleware):
+    def process_view(self, request, view_func, view_args, view_kwargs):
+        template = engines["django"].from_string(
+            "Processed view {{ view }}{% for m in mw %}\n{{ m }}{% endfor %}"
+        )
+        return TemplateResponse(
+            request,
+            template,
+            {"mw": [self.__class__.__name__], "view": view_func.__name__},
+        )
+
+
+class TemplateResponseMiddleware(BaseMiddleware):
+    def process_template_response(self, request, response):
+        response.context_data["mw"].append(self.__class__.__name__)
+        return response
+
+
+@async_only_middleware
+class AsyncTemplateResponseMiddleware(BaseMiddleware):
+    async def process_template_response(self, request, response):
+        response.context_data["mw"].append(self.__class__.__name__)
+        return response
+
+
+class LogMiddleware(BaseMiddleware):
+    async def __call__(self, request):
+        response = await self.get_response(request)
+        log.append((response.status_code, response.content))
+        return response
+
+
+class NoTemplateResponseMiddleware(BaseMiddleware):
+    def process_template_response(self, request, response):
+        return None
+
+
+@async_only_middleware
+class AsyncNoTemplateResponseMiddleware(BaseMiddleware):
+    async def process_template_response(self, request, response):
+        return None
+
+
+class NotFoundMiddleware(BaseMiddleware):
+    async def __call__(self, request):
+        raise Http404("not found")
+
+
+@async_only_middleware
+def async_payment_middleware(get_response):
+    async def middleware(request):
+        response = await get_response(request)
+        response.status_code = 402
+        return response
+
+    return middleware
+
+
+@sync_and_async_middleware
+class SyncAndAsyncMiddleware(BaseMiddleware):
+    pass
+
+
+class NotSyncOrAsyncMiddleware(BaseMiddleware):
+    """Middleware that is deliberately neither sync or async."""
+
+    sync_capable = False
+    async_capable = False
+
+    async def __call__(self, request):
+        return await self.get_response(request)