Writing Test Cases for API Route Views End Point

core/config.py

@@ -11,10 +11,12 @@ class Settings(BaseSettings):
     ALLOWED_HOSTS: str
     REFRESH_TOKEN_EXPIRE_DAYS: int
     ACCESS_TOKEN_EXPIRE_MINUTES: int
-    DEBUG: bool
+    DEVELOPMENT: bool
+    TESTING: bool
 
     # Databases
     POSTGRESQL_URL: str
+    SQLITE_TEST_URL: str
 
     # Media
     USER_AVATAR_PATH: str

core/versioning.py

@@ -30,7 +30,7 @@ def __init__(
             },
         }
 
-        if settings.DEBUG:
+        if settings.DEVELOPMENT:
             self.app = FastAPI(**__kwargs)
         else:
             self.app = FastAPI(docs_url=None, redoc_url=None, **__kwargs)

db/postgresql.py

@@ -5,8 +5,11 @@
 from core import settings
 
 
-database = Database(settings.POSTGRESQL_URL)
 metadata = MetaData()
+if not settings.TESTING:
+    database = Database(url=settings.POSTGRESQL_URL)
+else:
+    database = Database(url=settings.SQLITE_TEST_URL, force_rollback=True)
 
 
 class MainMeta(ModelMeta):
@@ -25,11 +28,11 @@ async def connect_to_postgresql():
     except Exception as e:
         logger.error(f"PostgreSQL Connection Failed {e}.")
     else:
-        logger.info("PostgreSQL Connected.")
+        logger.info(f"PostgreSQL Connected.")
 
 
 async def close_postgresql_connection():
     """Shutdown Event Handler for Disconnect to PostgreSQL Database"""
 
     await database.disconnect()
-    logger.info("PostgreSQL Closed.")
+    logger.info(f"PostgreSQL Closed.")

requirements.txt

@@ -6,6 +6,7 @@ argon2-cffi==21.3.0
 argon2-cffi-bindings==21.2.0
 asgiref==3.5.0
 asyncpg==0.25.0
+attrs==21.4.0
 bcrypt==3.2.0
 certifi==2021.10.8
 cffi==1.15.0
@@ -19,34 +20,45 @@ email-validator==1.1.3
 fastapi==0.75.0
 fastapi-utils==0.2.1
 greenlet==1.1.2
-h11==0.13.0
+h11==0.12.0
+httpcore==0.14.7
 httptools==0.2.0
+httpx==0.22.0
 idna==3.3
 importlib-metadata==4.11.3
 importlib-resources==5.4.0
+iniconfig==1.1.1
 itsdangerous==2.1.1
 Jinja2==3.1.0
 Mako==1.2.0
 MarkupSafe==2.1.1
 orjson==3.6.7
 orm==0.3.1
 ormar==0.10.25
+packaging==21.3
 passlib==1.7.4
+pluggy==1.0.0
 psycopg2-binary==2.9.3
+py==1.11.0
 pyasn1==0.4.8
 pycparser==2.21
 pydantic==1.9.0
+pyparsing==3.0.7
+pytest==7.1.1
+pytest-asyncio==0.18.3
 python-dotenv==0.19.2
 python-jose==3.3.0
 python-magic==0.4.25
 python-multipart==0.0.5
 PyYAML==5.4.1
 requests==2.27.1
+rfc3986==1.5.0
 rsa==4.8
 six==1.16.0
 sniffio==1.2.0
 SQLAlchemy==1.4.31
 starlette==0.17.1
+tomli==2.0.1
 typesystem==0.3.1
 typing-extensions==4.1.1
 ujson==4.3.0

setup.cfg

@@ -0,0 +1,5 @@
+[tool:pytest]
+asyncio_mode = auto
+filterwarnings =
+    ignore:.*U.*mode is deprecated:DeprecationWarning
+addopts = -p no:warnings

tests/api/v1_0_0/test_user.py

@@ -0,0 +1,123 @@
+from fastapi import status
+from httpx import AsyncClient
+from typing import Dict, Any
+from core import Level
+from ...conftest import VERSIONS, FIRST_ADMIN
+
+
+URL_STR: str = VERSIONS.get("1.0.0") + "user/"
+
+
+class TestUserRoutes:
+    """Test Cases Class for Test APIs of User Entity Model Routes"""
+
+    async def test_login_failed_username_user(
+        self,
+        client: AsyncClient,
+    ):
+        response = await client.post(
+            url=URL_STR + "login/",
+            data={
+                "username": "wrongusername",
+                "password": FIRST_ADMIN.get("password"),
+            },
+        )
+
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+
+    async def test_login_failed_password_user(
+        self,
+        client: AsyncClient,
+    ):
+        response = await client.post(
+            url=URL_STR + "login/",
+            data={
+                "username": FIRST_ADMIN.get("username"),
+                "password": "wrongpassword",
+            },
+        )
+
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+
+    async def test_login_successfull_user(
+        self,
+        client: AsyncClient,
+    ):
+        response = await client.post(url=URL_STR + "login/", data=FIRST_ADMIN)
+
+        assert response.status_code == status.HTTP_200_OK
+
+        content: Dict[str, Any] = response.json()
+
+        assert "token_type" in content
+        assert "access_token" in content
+        assert "refresh_token" in content
+
+    async def test_refresh_failed_user(
+        self,
+        client: AsyncClient,
+    ):
+        response = await client.post(
+            url=URL_STR + "refresh/",
+            json={
+                "refresh": "fakerefreshtoken",
+            },
+        )
+
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+
+    async def test_refresh_successfull_user(
+        self,
+        client: AsyncClient,
+    ):
+        response = await client.post(url=URL_STR + "login/", data=FIRST_ADMIN)
+        tokens: Dict[str, Any] = response.json()
+
+        response = await client.post(
+            url=URL_STR + "refresh/",
+            json={
+                "refresh": tokens.get("refresh_token"),
+            },
+        )
+
+        assert response.status_code == status.HTTP_200_OK
+
+        content: Dict[str, Any] = response.json()
+
+        assert "token_type" in content
+        assert "access_token" in content
+
+    async def test_list_user(
+        self,
+        client: AsyncClient,
+        admin_token_headers: Dict[str, str],
+    ):
+        response = await client.get(URL_STR, headers=admin_token_headers)
+
+        assert response.status_code == status.HTTP_200_OK
+
+        content: Dict[str, Any] = response.json()
+
+        assert content["count"] == 1
+        assert content["next"] is None
+        assert content["previous"] is None
+
+        assert content["results"][0]["mobile"] == FIRST_ADMIN.get("username")
+        assert content["results"][0]["level"] == Level.ADMIN.value
+        assert content["results"][0]["is_active"] is True
+
+    async def test_create_user(
+        self,
+        client: AsyncClient,
+        admin_token_headers: Dict[str, str],
+    ):
+        response = await client.post(
+            url=URL_STR,
+            headers=admin_token_headers,
+            json={
+                "mobile": "9123456789",
+                "password": "secretpassword",
+            },
+        )
+
+        assert response.status_code == status.HTTP_201_CREATED

tests/conftest.py

@@ -0,0 +1,86 @@
+from httpx import AsyncClient
+from pytest_asyncio import fixture
+from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
+from asyncio import get_event_loop
+from typing import Dict, Generator
+from core import settings, Level
+from db import metadata
+from models import User
+from schemas import UserInDBSchema
+from main import app
+
+
+engine: AsyncEngine = create_async_engine(settings.SQLITE_TEST_URL, echo=True)
+
+FIRST_ADMIN = {
+    "username": "9928917653",
+    "password": "sepehrbazyar",
+}
+
+VERSIONS = {
+    version: f"/v{version}/" for version in [
+        "1.0.0",
+    ]
+}
+
+
+@fixture(autouse=True, scope="session")
+async def create_test_database():
+    """SetUp & TearDown Fixture to Create & Delete Database by Metadata"""
+
+    async with engine.begin() as conn:
+        await conn.run_sync(metadata.create_all)
+        yield
+        await conn.run_sync(metadata.drop_all)
+
+
+@fixture(autouse=True, scope="function")
+async def clean_tables_database():
+    """Clearing Database Tables to Delete All Rows for Each Test Case"""
+
+    try:
+        yield
+    finally:
+        async with engine.begin() as conn:
+            for table in reversed(metadata.sorted_tables):
+                await conn.execute(table.delete())
+
+
+@fixture(scope="session")
+async def client() -> Generator[AsyncClient, None, None]:
+    """Session Generator Fixture to Yielding Async Client for Requesting APIs"""
+
+    async with AsyncClient(app=app, base_url=settings.BASE_URL) as client:
+        yield client
+
+
+@fixture(autouse=True, scope="function")
+async def insert_first_admin():
+    """Fixture to Inserting First Admin User for Access Protected Routes View"""
+
+    form = UserInDBSchema(
+        level=Level.ADMIN,
+        mobile=FIRST_ADMIN.get("username"),
+        password=FIRST_ADMIN.get("password"),
+    )
+    await User.sign_up(form=form)
+
+
+@fixture(scope="function")
+async def admin_token_headers(client: AsyncClient) -> Dict[str, str]:
+    """Sent Authorize Data to Login Super Admin Get Access Token"""
+
+    response = await client.post(f"/user/login/", data=FIRST_ADMIN)
+    tokens: Dict[str, str] = response.json()
+    return {
+        "Authorization": f"Bearer {tokens.get('access_token')}",
+    }
+
+
+@fixture(autouse=True, scope="session")
+def event_loop():
+    """Override the Event Loop Tests for Set Async Fixture Tests"""
+
+    loop = get_event_loop()
+    yield loop
+    loop.close()