feat: add full test

Author: Zxilly

examples/rbac_model.conf

@@ -11,4 +11,4 @@ g = _, _
 e = some(where (p.eft == allow))
 
 [matchers]
-m = (p.sub == "*" || g(r.sub, p.sub)) && r.obj == p.obj && (p.act == "*" || r.act == p.act)
+m = (p.sub == "*" || g(r.sub, p.sub)) && (r.obj == p.obj || keyMatch(r.obj, p.obj)) && (p.act == "*" || r.act == p.act)

fastapi_authz/middleware.py

@@ -1,12 +1,9 @@
-import typing
-
 from casbin.enforcer import Enforcer
-
+from starlette.authentication import BaseUser
 from starlette.requests import Request
 from starlette.responses import JSONResponse
 from starlette.status import HTTP_403_FORBIDDEN
 from starlette.types import ASGIApp, Receive, Scope, Send
-from starlette.authentication import BaseUser
 
 
 class CasbinMiddleware:
@@ -61,7 +58,10 @@ def _enforce(self, scope: Scope, receive: Receive) -> bool:
         if 'user' not in scope:
             raise RuntimeError("Casbin Middleware must work with an Authentication Middleware")
 
-        assert isinstance(request.user,BaseUser)
+        assert isinstance(request.user, BaseUser)
 
         user = request.user.display_name if request.user.is_authenticated else 'anonymous'
+
+        print(user, path, method)
+
         return self.enforcer.enforce(user, path, method)

tests/conftest.py

@@ -1,9 +1,9 @@
 import base64
 import binascii
+import os
 
 import casbin
 import pytest
-import os
 from fastapi import FastAPI
 from starlette.authentication import AuthenticationBackend, AuthenticationError, AuthCredentials, SimpleUser
 from starlette.middleware.authentication import AuthenticationMiddleware
@@ -38,9 +38,7 @@ def app_fixture():
 
     app = FastAPI()
 
-    @app.on_event('startup')
-    async def startup():
-        app.add_middleware(CasbinMiddleware, enforcer=enforcer)
-        app.add_middleware(AuthenticationMiddleware, backend=BasicAuth())
+    app.add_middleware(CasbinMiddleware, enforcer=enforcer)
+    app.add_middleware(AuthenticationMiddleware, backend=BasicAuth())
 
     yield app

tests/test_middleware.py

@@ -1,35 +1,41 @@
 import pytest
-from fastapi import Depends
-from starlette.authentication import requires
 from starlette.testclient import TestClient
 
 
 @pytest.mark.parametrize(
     "test_server_path, test_client_path, method, status_code, user, response_body", [
         ('/dataset1/resource2', '/dataset1/resource2', 'GET', 200, 'alice', 'ok'),
+        ('/dataset1/resource2', '/dataset1/resource2', 'GET', 403, 'notalice', 'Forbidden'),
+        ('/dataset1/resource1', '/dataset1/resource1', 'POST', 200, 'alice', 'ok'),
     ]
 )
 def test_middleware_authed(app_fixture, test_server_path, test_client_path, method, status_code, user, response_body):
-    # if method == 'GET':
-    #     @app_fixture.get(test_server_path)
-    #     async def index():
-    #         return 'ok'
-    # elif method == 'POST':
-    #     @app_fixture.post(test_server_path)
-    #     async def index():
-    #         return 'ok'
-    # elif method == 'PUT':
-    #     @app_fixture.put(test_server_path)
-    #     async def index():
-    #         return 'ok'
+    @getattr(app_fixture, method.lower())(test_server_path)
+    async def index():
+        return 'ok'
 
+    test_client = TestClient(app_fixture)
+
+    test_response = getattr(test_client, method.lower())(test_client_path, auth=(user, 'password'))
+
+    assert test_response.status_code == status_code
+    assert test_response.json() == response_body
+
+
+@pytest.mark.parametrize(
+    "test_server_path, test_client_path, method, status_code, response_body", [
+        ('/login', '/login', 'GET', 200, 'ok'),
+        ('/', '/', 'GET', 200, 'ok')
+    ]
+)
+def test_middleware_not_authed(app_fixture, test_server_path, test_client_path, method, status_code, response_body):
     @getattr(app_fixture, method.lower())(test_server_path)
     async def index():
         return 'ok'
 
     test_client = TestClient(app_fixture)
 
-    test_response = test_client.get(test_client_path, auth=(user, 'password'))
+    test_response = getattr(test_client, method.lower())(test_client_path)
 
     assert test_response.status_code == status_code
     assert test_response.json() == response_body