add uni tests for user auth

app/api/user.py

@@ -0,0 +1,34 @@
+from fastapi import APIRouter, Depends, status, Request, HTTPException
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.database import get_db
+from app.models.user import User
+from app.schemas.user import UserSchema, UserResponse, UserLogin, TokenResponse
+from app.services.auth import create_access_token
+
+router = APIRouter(prefix="/v1/user")
+
+
+@router.post("/", status_code=status.HTTP_201_CREATED, response_model=UserResponse)
+async def create_user(payload: UserSchema, request: Request, db_session: AsyncSession = Depends(get_db)):
+    _user: User = User(**payload.model_dump())
+    await _user.save(db_session)
+
+    # TODO: add refresh token
+    _user.access_token = await create_access_token(_user, request)
+    return _user
+
+
+@router.post("/token", status_code=status.HTTP_201_CREATED, response_model=TokenResponse)
+async def get_token_for_user(user: UserLogin, request: Request, db_session: AsyncSession = Depends(get_db)):
+    _user: User = await User.find(db_session, [User.email == user.email])
+
+    # TODO: out exception handling to external module
+    if not _user:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
+    if not _user.check_password(user.password):
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Password is incorrect")
+
+    # TODO: add refresh token
+    _token = await create_access_token(_user, request)
+    return {"access_token": _token, "token_type": "bearer"}

app/schemas/user.py

@@ -31,4 +31,3 @@ class UserLogin(BaseModel):
     model_config = config
     email: EmailStr = Field(title="User’s email", description="User’s email")
     password: str = Field(title="User’s password", description="User’s password")
-

tests/api/test_auth.py

@@ -0,0 +1,35 @@
+import pytest
+from httpx import AsyncClient
+from starlette import status
+import jwt
+
+pytestmark = pytest.mark.anyio
+
+# TODO: parametrize test with diff urls
+async def test_add_user(client: AsyncClient):
+    payload = {
+        "email": "rancher@grassroots.com",
+        "first_name": "Joe",
+        "last_name": "Garcia",
+        "password": "s1lly"
+    }
+    response = await client.post("/user/", json=payload)
+    assert response.status_code == status.HTTP_201_CREATED
+    claimset = jwt.decode(response.json()["access_token"], options={"verify_signature": False})
+    assert claimset["email"] == payload["email"]
+    assert claimset["expiry"] > 0
+    assert claimset["platform"] == "python-httpx/0.24.1"
+
+
+# TODO: parametrize test with diff urls including 404 and 401
+async def test_get_token(client: AsyncClient):
+    payload = {"email": "rancher@grassroots.com", "password": "s1lly"}
+    response = await client.post("/user/token", json=payload)
+    assert response.status_code == status.HTTP_201_CREATED
+    claimset = jwt.decode(response.json()["access_token"], options={"verify_signature": False})
+    assert claimset["email"] == payload["email"]
+    assert claimset["expiry"] > 0
+    assert claimset["platform"] == "python-httpx/0.24.1"
+
+
+# TODO: baerer token test > get token > test endpoint auth with token > expire token on redis > test endpoint auth with token

tests/api/test_health.py

@@ -0,0 +1,11 @@
+import pytest
+from fastapi import status
+from httpx import AsyncClient
+
+pytestmark = pytest.mark.anyio
+
+async def test_redis_health(client: AsyncClient):
+    response = await client.get(f"/public/health/redis")
+    assert response.status_code == status.HTTP_200_OK
+    # assert payload["name"] == response.json()["name"]
+    # assert UUID(response.json()["id"])

tests/conftest.py

@@ -1,13 +1,14 @@
 import pytest
-import pytest_asyncio
 from httpx import AsyncClient
 
 from app.database import engine
 from app.main import app
 from app.models.base import Base
+from app.redis import get_redis
 
 
 @pytest.fixture(
+    scope="session",
     params=[
         pytest.param(("asyncio", {"use_uvloop": True}), id="asyncio+uvloop"),
     ]
@@ -16,6 +17,7 @@ def anyio_backend(request):
     return request.param
 
 
+@pytest.fixture(scope="session")
 async def start_db():
     async with engine.begin() as conn:
         await conn.run_sync(Base.metadata.drop_all)
@@ -25,15 +27,12 @@ async def start_db():
     await engine.dispose()
 
 
-@pytest_asyncio.fixture
+@pytest.fixture(scope="session")
-async def client() -> AsyncClient:
+async def client(start_db) -> AsyncClient:
     async with AsyncClient(
         app=app,
         base_url="http://testserver/v1",
         headers={"Content-Type": "application/json"},
-    ) as client:
+    ) as test_client:
-        await start_db()
+        app.state.redis = await get_redis()
-        yield client
+        yield test_client
-        # for AsyncEngine created in function scope, close and
-        # clean-up pooled connections
-        await engine.dispose()