feat: add pytest

Author: Zxilly

examples/basic_policy.csv

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

examples/basic_model.conf examples/rbac_model.confexamples/basic_model.conf renamed to examples/rbac_model.conf

@@ -0,0 +1,11 @@
+p, alice, /dataset1/*, GET
+p, alice, /dataset1/resource1, POST
+p, bob, /dataset2/resource1, *
+p, bob, /dataset2/resource2, GET
+p, bob, /dataset2/folder1/*, POST
+p, dataset1_admin, /dataset1/*, *
+p, *, /login, *
+
+p, anonymous, /, GET
+
+g, cathy, dataset1_admin

examples/rbac_policy.csv

@@ -54,7 +54,7 @@
         "authorization",
         "permission"
     ],
-    packages=find_packages(exclude=["docs", "tests*"]),
+    packages=find_packages(exclude=["docs", "pytest*"]),
     data_files=[desc_file],
     include_package_data=True,
     install_requires=install_requires,

setup.py

@@ -0,0 +1 @@
+import pytest

tests/__init__.py

@@ -0,0 +1,46 @@
+import base64
+import binascii
+
+import casbin
+import pytest
+import os
+from fastapi import FastAPI
+from starlette.authentication import AuthenticationBackend, AuthenticationError, AuthCredentials, SimpleUser
+from starlette.middleware.authentication import AuthenticationMiddleware
+
+from fastapi_authz import CasbinMiddleware
+
+
+def get_examples(path):
+    examples_path = os.path.split(os.path.realpath(__file__))[0] + "/../examples/"
+    return os.path.abspath(examples_path + path)
+
+
+class BasicAuth(AuthenticationBackend):
+    async def authenticate(self, request):
+        if "Authorization" not in request.headers:
+            return None
+
+        auth = request.headers["Authorization"]
+        try:
+            scheme, credentials = auth.split()
+            decoded = base64.b64decode(credentials).decode("ascii")
+        except (ValueError, UnicodeDecodeError, binascii.Error):
+            raise AuthenticationError("Invalid basic auth credentials")
+
+        username, _, password = decoded.partition(":")
+        return AuthCredentials(["authenticated"]), SimpleUser(username)
+
+
+@pytest.fixture
+def app_fixture():
+    enforcer = casbin.Enforcer(get_examples("rbac_model.conf"), get_examples("rbac_policy.csv"))
+
+    app = FastAPI()
+
+    @app.on_event('startup')
+    async def startup():
+        app.add_middleware(CasbinMiddleware, enforcer=enforcer)
+        app.add_middleware(AuthenticationMiddleware, backend=BasicAuth())
+
+    yield app

tests/conftest.py

@@ -0,0 +1,39 @@
+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'),
+    ]
+)
+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 = test_client.get(test_client_path, auth=(user, 'password'))
+
+    assert test_response.status_code == status_code
+    assert test_response.json() == response_body
+
+
+if __name__ == '__main__':
+    pytest.main()