Add tests for authentication and utilities, and update dependencies

- Created a new test package for services and added tests for AuthService.
- Implemented tests for user registration, login, and token creation.
- Added a new test package for utilities and included tests for password and JWT utilities.
- Updated `uv.lock` to include new dependencies: bcrypt, email-validator, pyjwt, and pytest-asyncio.
This commit is contained in:
JSC
2025-07-25 17:48:43 +02:00
parent af20bc8724
commit e456d34897
23 changed files with 2381 additions and 8 deletions

View File

@@ -1,3 +1,5 @@
from typing import Literal
from pydantic_settings import BaseSettings, SettingsConfigDict
@@ -14,15 +16,27 @@ class Settings(BaseSettings):
HOST: str = "localhost"
PORT: int = 8000
RELOAD: bool = True
LOG_LEVEL: str = "info"
DATABASE_URL: str = "sqlite+aiosqlite:///data/soundboard.db"
DATABASE_ECHO: bool = False
LOG_LEVEL: str = "info"
LOG_FILE: str = "logs/app.log"
LOG_MAX_SIZE: int = 10 * 1024 * 1024
LOG_BACKUP_COUNT: int = 5
LOG_FORMAT: str = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
DATABASE_URL: str = "sqlite+aiosqlite:///data/soundboard.db"
DATABASE_ECHO: bool = False
# JWT Configuration
JWT_SECRET_KEY: str = (
"your-secret-key-change-in-production" # noqa: S105 default value if none set in .env
)
JWT_ALGORITHM: str = "HS256"
JWT_ACCESS_TOKEN_EXPIRE_MINUTES: int = 15 # Shorter-lived access token
JWT_REFRESH_TOKEN_EXPIRE_DAYS: int = 7 # Longer-lived refresh token
# Cookie Configuration
COOKIE_SECURE: bool = True # Set to False for development without HTTPS
COOKIE_SAMESITE: Literal["strict", "lax", "none"] = "lax"
settings = Settings()

103
app/core/dependencies.py Normal file
View File

@@ -0,0 +1,103 @@
"""FastAPI dependencies."""
from typing import Annotated, NoReturn, cast
from fastapi import Cookie, Depends, HTTPException, status
from sqlmodel.ext.asyncio.session import AsyncSession
from app.core.database import get_db
from app.core.logging import get_logger
from app.models.user import User
from app.services.auth import AuthService
from app.utils.auth import JWTUtils
logger = get_logger(__name__)
def _raise_invalid_token_error() -> NoReturn:
"""Raise an invalid token HTTP exception."""
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token payload",
)
def _raise_auth_error() -> NoReturn:
"""Raise an authentication HTTP exception."""
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
)
async def get_auth_service(
session: Annotated[AsyncSession, Depends(get_db)],
) -> AuthService:
"""Get the authentication service."""
return AuthService(session)
async def get_current_user(
access_token: Annotated[str | None, Cookie()],
auth_service: Annotated[AuthService, Depends(get_auth_service)],
) -> User:
"""Get the current authenticated user from JWT token in HTTP-only cookie."""
try:
# Check if access token cookie exists
if not access_token:
logger.warning("No access token cookie found")
_raise_auth_error()
# Decode the JWT token
payload = JWTUtils.decode_access_token(access_token)
# Extract user ID from token
user_id_str = payload.get("sub")
if not user_id_str:
_raise_invalid_token_error()
# At this point user_id_str is guaranteed to be truthy, safe to cast
user_id_str = cast("str", user_id_str)
try:
user_id = int(user_id_str)
except (ValueError, TypeError) as e:
logger.warning("Invalid user ID in token: %s", user_id_str)
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token payload",
) from e
# Get the user
return await auth_service.get_current_user(user_id)
except HTTPException:
# Re-raise HTTPExceptions without wrapping them
raise
except Exception:
logger.exception("Failed to authenticate user")
_raise_auth_error()
async def get_current_active_user(
current_user: Annotated[User, Depends(get_current_user)],
) -> User:
"""Get the current authenticated and active user."""
if not current_user.is_active:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Account is deactivated",
)
return current_user
async def get_admin_user(
current_user: Annotated[User, Depends(get_current_active_user)],
) -> User:
"""Get the current authenticated admin user."""
if current_user.role not in ["admin", "superadmin"]:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not enough permissions",
)
return current_user