Added create_admin tests

app/core/db.py

@@ -1,4 +1,4 @@
-from typing import Annotated, AsyncGenerator
+from typing import Annotated, Any, Generator
 
 from fastapi import Depends
 from sqlalchemy import MetaData
@@ -33,7 +33,7 @@ def __repr__(self) -> str:
 AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)
 
 
-async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
+async def get_async_session() -> Generator[AsyncSession, Any, None]:
     async with AsyncSessionLocal() as async_session:
         yield async_session
 

app/core/user.py

@@ -1,6 +1,6 @@
 import logging
-from contextlib import asynccontextmanager as acm
-from typing import Annotated
+from contextlib import asynccontextmanager
+from typing import Annotated, Generator
 
 from fastapi import Depends, Request
 from fastapi_users import BaseUserManager, FastAPIUsers, IntegerIDMixin
@@ -52,29 +52,24 @@ def get_jwt_strategy() -> JWTStrategy:
 authorized = Annotated[User, Depends(current_user)]
 admin = Annotated[User, Depends(current_superuser)]
 
-# Create superuseruser programmatically
-get_async_session_context = acm(get_async_session)
-get_user_db_context = acm(get_user_db)
-get_user_manager_context = acm(get_user_manager)
-
 
-async def create_user(email: EmailStr, password: str, is_superuser: bool = False):
+# Create superuseruser programmatically
+async def create_user(async_generator: Generator, email: EmailStr, password: str, is_superuser: bool = False) -> User:
     try:
-        async with get_async_session_context() as session:
-            async with get_user_db_context(session) as user_db:
-                async with get_user_manager_context(user_db) as user_manager:
-                    user = await user_manager.create(UserCreate(
-                        email=email,
-                        password=password,
-                        is_superuser=is_superuser))
-                    logger.info('Админ создан')
-                    return user
+        async with (asynccontextmanager(async_generator)() as session,
+                    asynccontextmanager(get_user_db)(session) as user_db,
+                    asynccontextmanager(get_user_manager)(user_db) as user_manager):
+            user = await user_manager.create(UserCreate(email=email, password=password, is_superuser=is_superuser))
     except UserAlreadyExists:
         logger.info('Админ уже существует')
+    else:
+        logger.info('Админ создан')
+        return user
 
 
-async def create_admin():
+async def create_admin(async_generator: Generator = get_async_session) -> User:
     if all((settings.admin_email, settings.admin_password)):
-        await create_user(email=settings.admin_email,
-                          password=settings.admin_password,
-                          is_superuser=True)
+        return await create_user(async_generator,
+                                 email=settings.admin_email,
+                                 password=settings.admin_password,
+                                 is_superuser=True)

tests/conftest.py

@@ -8,8 +8,9 @@
 from sqlalchemy.ext.asyncio import (AsyncSession, async_sessionmaker,
                                     create_async_engine)
 
-from app.core import (Base, current_user, get_aioredis, get_async_session,
-                      settings)
+from app.core import settings  # noqa
+from app.core import Base, current_user, get_aioredis, get_async_session
+from app.core.user import create_admin, create_user  # noqa
 from app.main import app
 from app.models import Post, User  # noqa
 from app.repositories.base_db_repository import CRUDBaseRepository  # noqa

tests/unit_tests/test_user.py

@@ -0,0 +1,24 @@
+import pytest
+
+from app.core.user import create_admin, create_user
+from tests.conftest import User, override_get_async_session, settings
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('is_superuser', (True, False))
+async def test_create_user(is_superuser) -> None:
+    user: User = await create_user(override_get_async_session,
+                                   settings.admin_email,
+                                   settings.admin_password,
+                                   is_superuser=is_superuser)
+    assert user.email == settings.admin_email
+    assert user.hashed_password
+    assert user.is_superuser == is_superuser
+
+
+@pytest.mark.asyncio
+async def test_create_admin() -> None:
+    user: User = await create_admin(override_get_async_session)
+    assert user.email == settings.admin_email
+    assert user.hashed_password
+    assert user.is_superuser