Writing Test Cases for API Route Views End Point #11

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft

SepehrBazyar wants to merge 30 commits into develop from test_apis

core/config.py

-Original file line number
+Diff line change
@@ Expand Up / @@ -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
@@ Expand Down @@

core/versioning.py

-Original file line number
+Diff line change
@@ Expand Up / @@ -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)
@@ Expand Down @@

db/postgresql.py

-Original file line number
+Diff line change
@@ Expand Up / @@ -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):
@@ Expand All / @@ -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

-Original file line number
+Diff line change
@@ Expand Up / @@ -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
@@ Expand All / @@ -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
@@ Expand Down @@

setup.cfg

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

tests/__init__.py

Empty file.

tests/api/__init__.py

Empty file.

tests/api/v1_0_0/__init__.py

Empty file.

tests/api/v1_0_0/test_user.py

-Original file line number
+Diff line change
@@ -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

-Original file line number
+Diff line change
@@ -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()

tests/models/__init__.py

Empty file.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Writing Test Cases for API Route Views End Point #11

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Writing Test Cases for API Route Views End Point #11

Are you sure you want to change the base?

Uh oh!

Writing Test Cases for API Route Views End Point #11

Uh oh!

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!