v0.2.0

Author: amisadmin

fastapi_user_auth/__init__.py

@@ -1,4 +1,4 @@
-__version__ = "0.1.3"
+__version__ = "0.2.0"
 __url__ = "https://github.com/amisadmin/fastapi_user_auth"
 
 import gettext

tests/conftest.py

@@ -0,0 +1,19 @@
+from sqlalchemy_database import Database, AsyncDatabase
+
+# sqlite
+sync_db = Database.create('sqlite:///amisadmin.db?check_same_thread=False')
+async_db = AsyncDatabase.create('sqlite+aiosqlite:///amisadmin.db?check_same_thread=False')
+
+# mysql
+# sync_db = Database.create('mysql+pymysql://root:[email protected]:3306/amisadmin?charset=utf8mb4')
+# async_db = AsyncDatabase.create('mysql+aiomysql://root:[email protected]:3306/amisadmin?charset=utf8mb4')
+
+# postgresql
+# sync_db = Database.create('postgresql://postgres:[email protected]:5432/amisadmin')
+# async_db = AsyncDatabase.create('postgresql+asyncpg://postgres:[email protected]:5432/amisadmin')
+
+# oracle
+# sync_db = Database.create('oracle+cx_oracle://scott:tiger@tnsname')
+
+# SQL Server
+# sync_db = Database.create('mssql+pyodbc://scott:tiger@mydsn')

tests/test_auth/conftest.py

@@ -1,8 +1,35 @@
+import asyncio
+from typing import AsyncGenerator
+from typing import Union
+
 import pytest
+import pytest_asyncio
+from fastapi import FastAPI
+from sqlalchemy.orm import Session
+from sqlalchemy_database import Database, AsyncDatabase
+from sqlmodel import SQLModel
 from starlette.testclient import TestClient
 
+from fastapi_user_auth.auth.auth import Auth, AuthRouter
+from fastapi_user_auth.auth.models import Role, Permission, Group
 from fastapi_user_auth.auth.models import User
-from tests.test_auth.main import app
+from tests.conftest import async_db, sync_db
+
+
+@pytest.fixture(params=[async_db, sync_db])
+async def db(request) -> Union[Database, AsyncDatabase]:
+    database = request.param
+    await database.async_run_sync(SQLModel.metadata.create_all, is_session=False)
+    yield database
+    await database.async_run_sync(SQLModel.metadata.drop_all, is_session=False)
+
+
+app = FastAPI()
+#  创建auth实例
+auth = Auth(db=async_db)
+#  注册auth基础路由
+auth_router = AuthRouter(auth=auth)
+app.include_router(auth_router.router)
 
 
 class UserClient:
@@ -26,7 +53,7 @@ def get_login_client(username: str = None, password: str = None) -> UserClient:
     return UserClient(client=client, user=user)
 
 
-@pytest.fixture(scope='session')
+@pytest.fixture
 def logins(request) -> UserClient:
     user_data = {
         'admin': {"username": "admin", "password": "admin"},
@@ -38,7 +65,51 @@ def logins(request) -> UserClient:
     return get_login_client(**user)
 
 
-@pytest.fixture(scope='session', autouse=True)
-def startup():
-    import asyncio
-    asyncio.run(app.router.startup())
+@pytest.fixture(scope="session")
+async def prepare_database() -> AsyncGenerator[None, None]:
+    await auth.db.async_run_sync(SQLModel.metadata.create_all, is_session=False)
+    yield
+    await auth.db.async_run_sync(SQLModel.metadata.drop_all, is_session=False)
+
+
+@pytest.fixture(scope="session")
+def event_loop():
+    loop = asyncio.get_event_loop_policy().new_event_loop()
+    yield loop
+    loop.close()
+
+
+@pytest_asyncio.fixture(scope="session", autouse=True)
+async def fake_users(prepare_database):
+    await auth.db.async_run_sync(create_fake_users)
+
+
+# noinspection PyTypeChecker
+def create_fake_users(session: Session):
+    # init permission
+    admin_perm = Permission(key='admin', name='admin permission')
+    vip_perm = Permission(key='vip', name='vip permission')
+    test_perm = Permission(key='test', name='test permission')
+    session.add_all([admin_perm, vip_perm, test_perm])
+    session.flush([admin_perm, vip_perm, test_perm])
+    # init role
+    admin_role = Role(key='admin', name='admin role', permissions=[admin_perm])
+    vip_role = Role(key='vip', name='vip role', permissions=[vip_perm])
+    test_role = Role(key='test', name='test role', permissions=[test_perm])
+    session.add_all([admin_role, vip_role, test_role])
+    session.flush([admin_role, vip_role, test_role])
+    # init group
+    admin_group = Group(key='admin', name='admin group', roles=[admin_role])
+    vip_group = Group(key='vip', name='vip group', roles=[vip_role])
+    test_group = Group(key='test', name='test group', roles=[test_role])
+    session.add_all([admin_group, vip_group, test_group])
+    session.flush([admin_group, vip_group, test_group])
+    # init user
+    admin_user = User(username='admin', password=auth.pwd_context.hash('admin'), email='[email protected]',
+                      roles=[admin_role], groups=[admin_group])
+    vip_user = User(username='vip', password=auth.pwd_context.hash('vip'), email='[email protected]', roles=[vip_role],
+                    groups=[vip_group])
+    test_user = User(username='test', password=auth.pwd_context.hash('test'), email='[email protected]', roles=[test_role],
+                     groups=[test_group])
+    session.add_all([admin_user, vip_user, test_user])
+    session.flush([admin_user, vip_user, test_user])