Added test for use global interceptor

Author: eadwinCode

ellar/core/guard/consumer.py

@@ -24,9 +24,6 @@ async def run_route_guards(self, context: IExecutionContext) -> None:
 
         for guard in self._get_guards(context):
             await self.run_guard(context, guard)
-        else:
-            for guard in map(self.get_guard_instance, context.get_app().get_guards()):
-                await self.run_guard(context, guard)
 
     async def run_guard(
         self, context: IExecutionContext, guard_instance: "GuardCanActivate"
@@ -43,7 +40,8 @@ def _get_guards(self, context: IExecutionContext) -> t.Iterable["GuardCanActivat
 
         return map(
             self.get_guard_instance,
-            reflector.get_all_and_override(GUARDS_KEY, *targets) or [],
+            reflector.get_all_and_override(GUARDS_KEY, *targets)
+            or context.get_app().get_guards(),
         )
 
     def get_guard_instance(

ellar/core/interceptors/consumer.py

@@ -28,14 +28,13 @@ def get_interceptor(
     async def execute(
         self, context: IExecutionContext, route_operation: "RouteOperationBase"
     ) -> t.Any:
-        interceptor_idx = 0
         route_interceptors: t.List[EllarInterceptor] = list(
             map(
                 self.get_interceptor,
                 self.reflector.get_all_and_override(
                     ROUTE_INTERCEPTORS, *[context.get_handler(), context.get_class()]
                 )
-                or [],
+                or context.get_app().get_interceptors(),
             )
         )
         route_interceptors_length = len(route_interceptors or [])
@@ -46,7 +45,7 @@ async def handler(idx: int) -> t.Any:
                 if idx >= route_interceptors_length:
                     return await route_operation.handle_request(context=context)
                 return await route_interceptors[idx].intercept(
-                    context, functools.partial(handler, interceptor_idx + 1)
+                    context, functools.partial(handler, idx + 1)
                 )
 
             res = await handler(0)

ellar/core/main.py

@@ -8,7 +8,7 @@
 from ellar.common.datastructures import State, URLPath
 from ellar.common.interfaces import IExceptionHandler, IExceptionMiddlewareService
 from ellar.common.logger import logger
-from ellar.common.models import GuardCanActivate
+from ellar.common.models import EllarInterceptor, GuardCanActivate
 from ellar.common.templating import Environment
 from ellar.common.types import ASGIApp, T, TReceive, TScope, TSend
 from ellar.core.middleware import (
@@ -54,6 +54,9 @@ def __init__(
         self._injector: EllarInjector = injector
 
         self._global_guards = [] if global_guards is None else list(global_guards)
+        self._global_interceptors: t.List[
+            t.Union[EllarInterceptor, t.Type[EllarInterceptor]]
+        ] = []
         self._exception_handlers = list(self.config.EXCEPTION_HANDLERS)
 
         self._user_middleware = list(t.cast(list, self.config.MIDDLEWARE))
@@ -143,11 +146,21 @@ def install_module(
     def get_guards(self) -> t.List[t.Union[t.Type[GuardCanActivate], GuardCanActivate]]:
         return self._global_guards
 
+    def get_interceptors(
+        self,
+    ) -> t.List[t.Union[EllarInterceptor, t.Type[EllarInterceptor]]]:
+        return self._global_interceptors
+
     def use_global_guards(
         self, *guards: t.Union["GuardCanActivate", t.Type["GuardCanActivate"]]
     ) -> None:
         self._global_guards.extend(guards)
 
+    def use_global_interceptors(
+        self, *interceptors: t.Union[EllarInterceptor, t.Type[EllarInterceptor]]
+    ) -> None:
+        self._global_interceptors.extend(interceptors)
+
     @property
     def injector(self) -> EllarInjector:
         return self._injector

tests/test_interceptors/test_interceptor.py

@@ -85,3 +85,25 @@ def test_interceptor_works_ws():
     with client.websocket_connect("/interceptor-ws") as ws:
         data = ws.receive_text()
     assert data == "intercepted okay"
+
+
+def test_global_guard_works():
+    tm = Test.create_test_module()
+    app = tm.create_application()
+
+    @get("/global")
+    def _interceptor_demo_endpoint():
+        return {"message": "intercepted okay"}
+
+    app.router.append(_interceptor_demo_endpoint)
+    app.use_global_interceptors(Interceptor1(), InterceptCustomException)
+
+    _client = tm.get_test_client()
+
+    res = _client.get("/global")
+
+    assert res.status_code == 200
+    assert res.json() == {
+        "Interceptor1": "Interceptor1 modified returned resulted",
+        "message": "intercepted okay",
+    }