Passing unit tests for models.py

Author: chriscarrollsmith

main.py

@@ -6,13 +6,17 @@
 from fastapi.staticfiles import StaticFiles
 from fastapi.templating import Jinja2Templates
 from fastapi.exceptions import RequestValidationError, StarletteHTTPException
-from routers import authentication, organization, role, user, dashboard, terms_of_service, privacy_policy, about
-from utils.auth import (
-    NeedsNewTokens,
-    PasswordValidationError,
-    AuthenticationError,
+from routers import about, account, dashboard, organization, privacy_policy, role, terms_of_service, user
+from utils.dependencies import (
     get_optional_user
 )
+from exceptions.http_exceptions import (
+    AuthenticationError,
+    PasswordValidationError
+)
+from exceptions.exceptions import (
+    NeedsNewTokens
+)
 from utils.db import set_up_db
 from utils.models import User
 
@@ -39,14 +43,14 @@ async def lifespan(app: FastAPI):
 # --- Include Routers ---
 
 
-app.include_router(authentication.router)
+app.include_router(account.router)
+app.include_router(about.router)
+app.include_router(dashboard.router)
 app.include_router(organization.router)
+app.include_router(privacy_policy.router)
 app.include_router(role.router)
-app.include_router(user.router)
-app.include_router(dashboard.router)
 app.include_router(terms_of_service.router)
-app.include_router(privacy_policy.router)
-app.include_router(about.router)
+app.include_router(user.router)
 
 
 # --- Exception Handling Middlewares ---

routers/about.py

@@ -1,7 +1,7 @@
 from typing import Optional
 from fastapi import APIRouter, Depends, Request
 from fastapi.templating import Jinja2Templates
-from utils.auth import get_optional_user
+from utils.dependencies import get_optional_user
 from utils.models import User
 
 router = APIRouter(prefix="/about", tags=["about"])

routers/dashboard.py

@@ -1,7 +1,7 @@
 from typing import Optional
 from fastapi import APIRouter, Depends, Request
 from fastapi.templating import Jinja2Templates
-from utils.auth import get_user_with_relations
+from utils.dependencies import get_user_with_relations
 from utils.models import User
 
 router = APIRouter(prefix="/dashboard", tags=["dashboard"])

routers/organization.py

@@ -5,9 +5,10 @@
 from fastapi.templating import Jinja2Templates
 from pydantic import BaseModel, ConfigDict, field_validator
 from sqlmodel import Session, select
-from utils.db import get_session
-from utils.auth import get_authenticated_user, get_user_with_relations
-from utils.models import Organization, User, Role, utc_time, default_roles, ValidPermissions
+from utils.db import get_session, default_roles
+from utils.dependencies import get_authenticated_user, get_user_with_relations
+from utils.models import Organization, User, Role, utc_time
+from utils.enums import ValidPermissions
 from exceptions.http_exceptions import EmptyOrganizationNameError, OrganizationNotFoundError, OrganizationNameTakenError, InsufficientPermissionsError
 
 logger = getLogger("uvicorn.error")

routers/privacy_policy.py

@@ -1,7 +1,7 @@
 from typing import Optional
 from fastapi import APIRouter, Depends, Request
 from fastapi.templating import Jinja2Templates
-from utils.auth import get_optional_user
+from utils.dependencies import get_optional_user
 from utils.models import User
 
 router = APIRouter(prefix="/privacy_policy", tags=["privacy_policy"])

routers/role.py

@@ -8,7 +8,7 @@
 from sqlmodel import Session, select, col
 from sqlalchemy.orm import selectinload
 from utils.db import get_session
-from utils.auth import get_authenticated_user
+from utils.dependencies import get_authenticated_user
 from utils.models import Role, Permission, ValidPermissions, utc_time, User, DataIntegrityError
 from exceptions.http_exceptions import InsufficientPermissionsError, InvalidPermissionError, RoleAlreadyExistsError, RoleNotFoundError, RoleHasUsersError
 

routers/terms_of_service.py

@@ -1,7 +1,7 @@
 from typing import Optional
 from fastapi import APIRouter, Depends, Request
 from fastapi.templating import Jinja2Templates
-from utils.auth import get_optional_user
+from utils.dependencies import get_optional_user
 from utils.models import User
 
 router = APIRouter(prefix="/terms_of_service", tags=["terms_of_service"])

routers/user.py

@@ -4,7 +4,8 @@
 from typing import Optional
 from fastapi.templating import Jinja2Templates
 from utils.models import UserBase, User, DataIntegrityError
-from utils.auth import get_session, get_authenticated_user
+from utils.db import get_session
+from utils.dependencies import get_authenticated_user
 from utils.images import validate_and_process_image, MAX_FILE_SIZE, MIN_DIMENSION, MAX_DIMENSION, ALLOWED_CONTENT_TYPES
 
 router = APIRouter(prefix="/user", tags=["user"])

tests/conftest.py

@@ -1,16 +1,14 @@
 import pytest
 from typing import Generator
-from dotenv import load_dotenv
-from sqlmodel import create_engine, Session, select
+from sqlmodel import create_engine, Session, select, SQLModel
 from sqlalchemy import Engine
 from fastapi.testclient import TestClient
-from utils.db import get_connection_url, set_up_db, tear_down_db, get_session
-from utils.models import User, PasswordResetToken, Organization, Role, UserPassword
+from utils.db import get_session
+from utils.models import User, PasswordResetToken, Organization, Role, Account, Permission
 from utils.auth import get_password_hash, create_access_token, create_refresh_token
+from utils.enums import ValidPermissions
 from main import app
 
-load_dotenv()
-
 
 # Define a custom exception for test setup errors
 class SetupError(Exception):
@@ -26,9 +24,8 @@ def engine() -> Engine:
     Create a new SQLModel engine for the test database.
     Use an in-memory SQLite database for testing.
     """
-    engine = create_engine(
-        get_connection_url()
-    )
+    # Use in-memory SQLite for testing
+    engine = create_engine("sqlite:///:memory:")
     return engine
 
 
@@ -38,9 +35,22 @@ def set_up_database(engine) -> Generator[None, None, None]:
     Set up the test database before running the test suite.
     Drop all tables and recreate them to ensure a clean state.
     """
-    set_up_db(drop=True)
+    # Create all tables in the in-memory database
+    SQLModel.metadata.create_all(engine)
+    
+    # Create permissions
+    with Session(engine) as session:
+        # Check if permissions already exist
+        existing_permissions = session.exec(select(Permission)).all()
+        if not existing_permissions:
+            # Create all permissions from the ValidPermissions enum
+            for permission in ValidPermissions:
+                session.add(Permission(name=permission))
+            session.commit()
+    
     yield
-    tear_down_db()
+    # Drop all tables
+    SQLModel.metadata.drop_all(engine)
 
 
 @pytest.fixture
@@ -57,22 +67,37 @@ def clean_db(session: Session) -> None:
     """
     Cleans up the database tables before each test.
     """
-    for model in (PasswordResetToken, User, Role, Organization):
+    # Don't delete permissions as they are required for tests
+    for model in (PasswordResetToken, User, Role, Organization, Account):
         for record in session.exec(select(model)).all():
             session.delete(record)
 
     session.commit()
 
 
 @pytest.fixture()
-def test_user(session: Session) -> User:
+def test_account(session: Session) -> Account:
+    """
+    Creates a test account in the database.
+    """
+    account = Account(
+        email="[email protected]",
+        hashed_password=get_password_hash("Test123!@#")
+    )
+    session.add(account)
+    session.commit()
+    session.refresh(account)
+    return account
+
+
+@pytest.fixture()
+def test_user(session: Session, test_account: Account) -> User:
     """
     Creates a test user in the database.
     """
     user = User(
         name="Test User",
-        email="[email protected]",
-        password=UserPassword(hashed_password=get_password_hash("Test123!@#"))
+        account_id=test_account.id
     )
     session.add(user)
     session.commit()
@@ -95,7 +120,7 @@ def get_session_override():
 
 
 @pytest.fixture()
-def auth_client(session: Session, test_user: User) -> Generator[TestClient, None, None]:
+def auth_client(session: Session, test_account: Account) -> Generator[TestClient, None, None]:
     """
     Provides a TestClient instance with valid authentication tokens.
     """
@@ -106,8 +131,8 @@ def get_session_override():
     client = TestClient(app)
 
     # Create and set valid tokens
-    access_token = create_access_token({"sub": test_user.email})
-    refresh_token = create_refresh_token({"sub": test_user.email})
+    access_token = create_access_token({"sub": test_account.email})
+    refresh_token = create_refresh_token({"sub": test_account.email})
 
     client.cookies.set("access_token", access_token)
     client.cookies.set("refresh_token", refresh_token)

tests/test_models.py

@@ -9,7 +9,8 @@
     ValidPermissions,
     User,
     UserRoleLink,
-    PasswordResetToken
+    PasswordResetToken,
+    Account
 )
 from .conftest import SetupError
 
@@ -139,13 +140,13 @@ def test_cascade_delete_organization(session: Session, test_user: User, test_org
     assert remaining_user.id == test_user.id
 
 
-def test_password_reset_token_cascade_delete(session: Session, test_user: User):
+def test_password_reset_token_cascade_delete(session: Session, test_account: Account):
     """
-    Test that password reset tokens are deleted when a user is deleted
+    Test that password reset tokens are deleted when an account is deleted
     """
-    # Create reset tokens for the user
-    token1 = PasswordResetToken(user_id=test_user.id)
-    token2 = PasswordResetToken(user_id=test_user.id)
+    # Create reset tokens for the account
+    token1 = PasswordResetToken(account_id=test_account.id)
+    token2 = PasswordResetToken(account_id=test_account.id)
     session.add(token1)
     session.add(token2)
     session.commit()
@@ -154,29 +155,29 @@ def test_password_reset_token_cascade_delete(session: Session, test_user: User):
     tokens = session.exec(select(PasswordResetToken)).all()
     assert len(tokens) == 2
 
-    # Delete the user
-    session.delete(test_user)
+    # Delete the account
+    session.delete(test_account)
     session.commit()
 
     # Verify tokens were cascade deleted
     remaining_tokens = session.exec(select(PasswordResetToken)).all()
     assert len(remaining_tokens) == 0
 
 
-def test_password_reset_token_is_expired(session: Session, test_user: User):
+def test_password_reset_token_is_expired(session: Session, test_account: Account):
     """
     Test that password reset token expiration is properly set and checked
     """
     # Create an expired token
     expired_token = PasswordResetToken(
-        user_id=test_user.id,
+        account_id=test_account.id,
         expires_at=datetime.now(UTC) - timedelta(hours=1)
     )
     session.add(expired_token)
 
     # Create a valid token
     valid_token = PasswordResetToken(
-        user_id=test_user.id,
+        account_id=test_account.id,
         expires_at=datetime.now(UTC) + timedelta(hours=1)
     )
     session.add(valid_token)
@@ -228,3 +229,18 @@ def test_user_has_permission(session: Session, test_user: User, test_organizatio
         ValidPermissions.EDIT_ORGANIZATION, test_organization) is True
     assert test_user.has_permission(
         ValidPermissions.INVITE_USER, test_organization) is False
+
+
+def test_cascade_delete_account_deletes_user(session: Session, test_account: Account, test_user: User):
+    """
+    Test that deleting an account cascades to delete the associated user
+    """
+    # Verify the user exists
+    assert session.exec(select(User).where(User.account_id == test_account.id)).first() is not None
+    
+    # Delete the account
+    session.delete(test_account)
+    session.commit()
+    
+    # Verify the user was cascade deleted
+    assert session.exec(select(User).where(User.account_id == test_account.id)).first() is None