Auto-PR: Dev to Master

app/database.py

@@ -4,7 +4,7 @@
 
 from app.settings import Settings
 
-settings = Settings()  # noqa
+settings = Settings()  # type: ignore
 
 # Add Support for Both ASYNC and SYNC Database URLs
 # With Async being the center of focus

app/routers/tests/test_auth.py

@@ -1,7 +1,6 @@
 import pytest
-from httpx import ASGITransport, AsyncClient, Response
+from httpx import AsyncClient, Response
 
-from app.main import app
 from app.models import User as UserDB
 from app.redis_manager import redis_manager
 from app.schemas.auth import UserModel
@@ -53,21 +52,6 @@ async def test_signup_fails(
     assert error_msg == error_message
 
 
-@pytest.fixture
-async def sign_up_user(signup_data: dict):
-    # to be used as a side effect to test things that need a user that has signed up
-    async with AsyncClient(
-        transport=ASGITransport(app=app), base_url="http://test"
-    ) as client:
-        await client.post("/v1/auth/signup", json=signup_data)
-
-
-async def mutate_cache_item(key: str, value: dict):
-    # to be used as a side effect to mutate cache item
-
-    redis_manager.cache_json_item(key, value)
-
-
 async def test_initiate_password_reset(client: AsyncClient, signup_data: dict):
     data = {"email": signup_data["email"]}
     response: Response = await client.post(
@@ -106,19 +90,51 @@ async def test_reset_password_fails(client: AsyncClient, user: UserDB):
     assert response.json().get("detail") == "Invalid Reset Code"
 
 
-async def test_signin(client: AsyncClient, user: UserDB):
-    data = {
-        "username": user.email,
-        "password": "password",
-    }
-    response: Response = await client.post("/v1/auth/token", data=data)
-    assert response.status_code == 200
-    response_data = response.json()
-    assert "token_type" in response_data
-    assert "access_token" in response_data
-    assert "refresh_token" in response_data
-    assert "access_expires_at" in response_data
-    assert "refresh_expires_at" in response_data
+class TestSignIn:
+    async def test_signin_success(self, client: AsyncClient, user: UserDB):
+        data = {
+            "username": user.email,
+            "password": "password",
+        }
+        response = await client.post("/v1/auth/token", data=data)
+        assert response.status_code == 200
+        response_data = response.json()
+        assert "access_token" in response_data
+        assert "token_type" in response_data
+        assert "refresh_token" in response_data
+        assert "access_expires_at" in response_data
+        assert "refresh_expires_at" in response_data
+        assert response_data["token_type"] == "Bearer"
+
+    async def test_signin_validation_error(self, client: AsyncClient):
+        # Pass a string that is NOT a valid email
+        data = {
+            "username": "not-an-email",
+            "password": "password",
+        }
+        response = await client.post("/v1/auth/token", data=data)
+
+        assert response.status_code == 422
+        assert "detail" in response.json()
+
+    async def test_signin_invalid_password(self, client: AsyncClient, user: UserDB):
+        data = {
+            "username": user.email,
+            "password": "wrong-password",
+        }
+        response = await client.post("/v1/auth/token", data=data)
+
+        # auth_services likely raises 401 for wrong passwords
+        assert response.status_code == 401
+        assert response.json()["detail"] == "Incorrect email or password"
+
+    async def test_signin_user_not_found(self, client: AsyncClient):
+        data = {
+            "username": "ghost@example.com",
+            "password": "password",
+        }
+        response = await client.post("/v1/auth/token", data=data)
+        assert response.status_code == 401
 
 
 async def test_get_refresh_token(client: AsyncClient, user: UserDB):

app/schemas/tests/test_auth.py

@@ -0,0 +1,22 @@
+import pytest
+from pydantic import ValidationError
+
+from app.schemas import auth as auth_schemas
+
+
+def test_update_user_model_succeeds():
+    auth_schemas.UpdateUserModel(
+        old_password="password",
+        new_password="newpassword",
+    )
+
+
+def test_update_user_model_fails():
+    with pytest.raises(ValidationError) as err:
+        auth_schemas.UpdateUserModel(
+            new_password="newpassword",
+        )
+
+    assert "old_password is required if new_password is provided." in str(
+        err.value.errors()
+    )

app/services/auth.py

@@ -81,7 +81,7 @@ async def verify_user(
         logger.info(f"Corrupt Log Data: {data}")
         raise HTTPException(status_code=404, detail="Invalid Verification Code")
 
-    if not data or data.get("code") != verification_data.code:
+    if data.get("code") != verification_data.code:
         logger.info("Invalid Verification Code")
         raise HTTPException(status_code=400, detail="Invalid Verification Code")
 
@@ -143,7 +143,7 @@ async def initiate_password_reset(
         send_mail,
         subject="Password Reset",
         receipients=[user.email],
-        payload={"username": user.username.title(), "code": code},
+        payload={"username": user.email.split("@")[0], "code": code},
         template="auth/initiate_password_reset.html",
     )
 

app/services/tests/test_auth.py

@@ -1,5 +1,8 @@
+from typing import Any
+
+import pytest
 from faker import Faker
-from fastapi import BackgroundTasks
+from fastapi import BackgroundTasks, HTTPException
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.models import User as UserDB
@@ -32,7 +35,17 @@ async def test_create_user(session: AsyncSession, signup_data: dict[str, str]):
     assert isinstance(result, UserDB)
 
 
-async def test_verify_user(user: UserDB, session: AsyncSession):
+async def test_create_user_fails(
+    session: AsyncSession, signup_data: dict[str, Any], user: UserDB  # noqa
+):  # noqa
+    signup_data["email"] = user.email
+    with pytest.raises(HTTPException) as err:
+        await auth_services.create_user(UserSignUpData(**signup_data), session)
+
+    assert "Email already registered" in str(err.value)
+
+
+async def test_verify_user_succeeds(user: UserDB, session: AsyncSession):
     verification_data = auth_schemas.UserVerificationModel(
         email=user.email, code="000000"
     )
@@ -45,3 +58,31 @@ async def test_verify_user(user: UserDB, session: AsyncSession):
         session,
     )
     # test that the flag for verified user is activated
+
+
+@pytest.mark.parametrize(
+    "verification_data,cached_code",
+    [
+        ({"email": faker.email(), "code": "000000"}, {"code": "000000"}),
+        ({"email": faker.email(), "code": "000000"}, {"code": "000001"}),
+        ({"email": faker.email(), "code": "000000"}, None),
+    ],
+)
+async def test_verify_user_fails(
+    user: UserDB,
+    session: AsyncSession,
+    verification_data: dict[str, str],
+    cached_code: dict[str, str] | None,
+):
+    data = auth_schemas.UserVerificationModel(**verification_data)
+    redis_manager.cache_json_item(
+        f"verification-code-{data.email}", cached_code  # type: ignore
+    )
+    with pytest.raises(HTTPException) as err:
+        await auth_services.verify_user(
+            data,
+            BackgroundTasks(tasks=[]),
+            session,
+        )
+
+    assert err.value.detail == "Invalid Verification Code"