Correct association table relationships to remove overlaps and silence warnings

Author: chriscarrollsmith

tests/conftest.py

@@ -1,6 +1,6 @@
 import pytest
 from dotenv import load_dotenv
-from sqlmodel import create_engine, Session, delete
+from sqlmodel import create_engine, Session, select
 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
@@ -47,9 +47,15 @@ def clean_db(session: Session):
     """
     Cleans up the database tables before each test.
     """
-    # Exempt from mypy until SQLModel overload properly supports delete()
-    session.exec(delete(PasswordResetToken))  # type: ignore
-    session.exec(delete(User))  # type: ignore
+    # Delete all PasswordResetTokens
+    tokens = session.exec(select(PasswordResetToken)).all()
+    for token in tokens:
+        session.delete(token)
+
+    # Delete all Users
+    users = session.exec(select(User)).all()
+    for user in users:
+        session.delete(user)
 
     session.commit()
 

tests/test_db.py

@@ -9,7 +9,7 @@
     set_up_db,
 )
 from utils.models import Role, Permission, Organization, RolePermissionLink, ValidPermissions
-from sqlalchemy import create_engine
+from sqlalchemy import Engine
 
 
 def test_get_connection_url():
@@ -126,7 +126,7 @@ def test_assign_permissions_to_role_duplicate_check(session: Session, test_organ
     assert len(link_count) == 1
 
 
-def test_set_up_db_creates_tables():
+def test_set_up_db_creates_tables(engine: Engine, session: Session):
     """Test that set_up_db creates all expected tables without warnings"""
     # First tear down any existing tables
     tear_down_db()
@@ -135,7 +135,6 @@ def test_set_up_db_creates_tables():
     set_up_db(drop=False)
 
     # Use SQLAlchemy inspect to check tables
-    engine = create_engine(get_connection_url())
     inspector = inspect(engine)
     table_names = inspector.get_table_names()
 
@@ -145,45 +144,33 @@ def test_set_up_db_creates_tables():
         "organization",
         "role",
         "permission",
-        "role_permission_link",
-        "password_reset_token"
+        "rolepermissionlink",
+        "passwordresettoken"
     }
     assert expected_tables.issubset(set(table_names))
 
     # Verify permissions were created
-    with Session(engine) as session:
-        permissions = session.exec(select(Permission)).all()
-        assert len(permissions) == len(ValidPermissions)
-
-    # Clean up
-    tear_down_db()
-    engine.dispose()
+    permissions = session.exec(select(Permission)).all()
+    assert len(permissions) == len(ValidPermissions)
 
 
-def test_set_up_db_drop_flag():
+def test_set_up_db_drop_flag(engine: Engine, session: Session):
     """Test that set_up_db's drop flag properly recreates tables"""
     # Set up db with drop=True
-    engine = create_engine(get_connection_url())
     set_up_db(drop=True)
 
-    # Create a new session for this test
-    with Session(engine) as session:
-        # Verify valid permissions exist
-        permissions = session.exec(select(Permission)).all()
-        assert len(permissions) == len(ValidPermissions)
-
-        # Create an organization
-        org = Organization(name="Test Organization")
-        session.add(org)
-        session.commit()
+    # Verify valid permissions exist
+    permissions = session.exec(select(Permission)).all()
+    assert len(permissions) == len(ValidPermissions)
 
-        # Set up db with drop=False
-        set_up_db(drop=False)
+    # Create an organization
+    org = Organization(name="Test Organization")
+    session.add(org)
+    session.commit()
 
-        # Verify organization exists
-        assert session.exec(select(Organization).where(
-            Organization.name == "Test Organization")).first() is not None
+    # Set up db with drop=False
+    set_up_db(drop=False)
 
-    # Clean up
-    tear_down_db()
-    engine.dispose()
+    # Verify organization exists
+    assert session.exec(select(Organization).where(
+        Organization.name == "Test Organization")).first() is not None

tests/test_models.py

@@ -12,44 +12,53 @@
 def test_permissions_persist_after_role_deletion(session: Session):
     """
     Test that permissions are not deleted when a related Role is deleted.
+    Permissions links are automatically deleted due to cascade_delete=True.
     """
+    # Verify all ValidPermissions exist in database
+    all_permissions = session.exec(select(Permission)).all()
+    assert len(all_permissions) == len(ValidPermissions)
+
     # Create an organization
     organization = Organization(name="Test Organization")
     session.add(organization)
     session.commit()
     session.refresh(organization)
 
-    # Create permissions
-    permission1 = Permission(name=ValidPermissions.DELETE_ORGANIZATION)
-    permission2 = Permission(name=ValidPermissions.EDIT_ORGANIZATION)
-    session.add_all([permission1, permission2])
-    session.commit()
-
-    # Create a role and link permissions
+    # Create a role and link two specific permissions
     role = Role(name="Test Role", organization_id=organization.id)
     session.add(role)
     session.commit()
     session.refresh(role)
 
+    # Find specific permissions to link
+    delete_org_permission = next(
+        p for p in all_permissions if p.name == ValidPermissions.DELETE_ORGANIZATION)
+    edit_org_permission = next(
+        p for p in all_permissions if p.name == ValidPermissions.EDIT_ORGANIZATION)
+
     role_permission_link1 = RolePermissionLink(
-        role_id=role.id, permission_id=permission1.id
+        role_id=role.id, permission_id=delete_org_permission.id
     )
     role_permission_link2 = RolePermissionLink(
-        role_id=role.id, permission_id=permission2.id
+        role_id=role.id, permission_id=edit_org_permission.id
     )
     session.add_all([role_permission_link1, role_permission_link2])
     session.commit()
 
-    # Delete the role
+    # Verify that RolePermissionLinks exist before deletion
+    role_permissions = session.exec(select(RolePermissionLink)).all()
+    assert len(role_permissions) == 2
+
+    # Delete the role (this will cascade delete the permission links)
     session.delete(role)
     session.commit()
 
-    # Verify that permissions still exist
+    # Verify that all permissions still exist
     remaining_permissions = session.exec(select(Permission)).all()
-    assert len(remaining_permissions) == 2
-    assert permission1 in remaining_permissions
-    assert permission2 in remaining_permissions
+    assert len(remaining_permissions) == len(ValidPermissions)
+    assert delete_org_permission in remaining_permissions
+    assert edit_org_permission in remaining_permissions
 
-    # Verify that RolePermissionLinks are deleted
+    # Verify that RolePermissionLinks were cascade deleted
     remaining_role_permissions = session.exec(select(RolePermissionLink)).all()
     assert len(remaining_role_permissions) == 0

utils/models.py

@@ -3,7 +3,7 @@
 from datetime import datetime, UTC, timedelta
 from typing import Optional, List
 from sqlmodel import SQLModel, Field, Relationship
-from sqlalchemy import Column, Enum as SQLAlchemyEnum
+from sqlalchemy import Column, Enum as SQLAlchemyEnum, ForeignKey
 
 
 def utc_time():
@@ -26,16 +26,21 @@ class ValidPermissions(Enum):
 
 class UserOrganizationLink(SQLModel, table=True):
     id: Optional[int] = Field(default=None, primary_key=True)
-    user_id: int = Field(foreign_key="user.id", ondelete="CASCADE")
-    organization_id: int = Field(
-        foreign_key="organization.id", ondelete="CASCADE")
+    user_id: int = Field(foreign_key="user.id")
+    organization_id: int = Field(foreign_key="organization.id")
     role_id: int = Field(foreign_key="role.id")
     created_at: datetime = Field(default_factory=utc_time)
     updated_at: datetime = Field(default_factory=utc_time)
 
-    user: "User" = Relationship(back_populates="organization_links")
-    organization: "Organization" = Relationship(back_populates="user_links")
-    role: "Role" = Relationship(back_populates="user_links")
+    user: "User" = Relationship(
+        back_populates="organization_links"
+    )
+    organization: "Organization" = Relationship(
+        back_populates="user_links"
+    )
+    role: "Role" = Relationship(
+        back_populates="user_links"
+    )
 
 
 class RolePermissionLink(SQLModel, table=True):
@@ -46,17 +51,35 @@ class RolePermissionLink(SQLModel, table=True):
     updated_at: datetime = Field(default_factory=utc_time)
 
 
+class Permission(SQLModel, table=True):
+    """
+    Represents a permission that can be assigned to a role. Should not be
+    modified unless the application logic and ValidPermissions enum change.
+    """
+    id: Optional[int] = Field(default=None, primary_key=True)
+    name: ValidPermissions = Field(
+        sa_column=Column(SQLAlchemyEnum(ValidPermissions, create_type=False)))
+    created_at: datetime = Field(default_factory=utc_time)
+    updated_at: datetime = Field(default_factory=utc_time)
+
+    roles: List["Role"] = Relationship(
+        back_populates="permissions",
+        link_model=RolePermissionLink
+    )
+
+
 class Organization(SQLModel, table=True):
     id: Optional[int] = Field(default=None, primary_key=True)
     name: str
     created_at: datetime = Field(default_factory=utc_time)
     updated_at: datetime = Field(default_factory=utc_time)
 
     user_links: List[UserOrganizationLink] = Relationship(
-        back_populates="organization")
-    users: List["User"] = Relationship(
-        back_populates="organizations",
-        link_model=UserOrganizationLink
+        back_populates="organization",
+        sa_relationship_kwargs={
+            "cascade": "all, delete-orphan",
+            "passive_deletes": True
+        }
     )
     roles: List["Role"] = Relationship(back_populates="organization")
 
@@ -87,25 +110,9 @@ class Role(SQLModel, table=True):
     )
 
 
-class Permission(SQLModel, table=True):
-    """
-    Represents a permission that can be assigned to a role.
-    """
-    id: Optional[int] = Field(default=None, primary_key=True)
-    name: ValidPermissions = Field(
-        sa_column=Column(SQLAlchemyEnum(ValidPermissions, create_type=False)))
-    created_at: datetime = Field(default_factory=utc_time)
-    updated_at: datetime = Field(default_factory=utc_time)
-
-    roles: List["Role"] = Relationship(
-        back_populates="permissions",
-        link_model=RolePermissionLink
-    )
-
-
 class PasswordResetToken(SQLModel, table=True):
     id: Optional[int] = Field(default=None, primary_key=True)
-    user_id: Optional[int] = Field(foreign_key="user.id", ondelete="CASCADE")
+    user_id: Optional[int] = Field(foreign_key="user.id")
     token: str = Field(default_factory=lambda: str(
         uuid4()), index=True, unique=True)
     expires_at: datetime = Field(
@@ -116,6 +123,7 @@ class PasswordResetToken(SQLModel, table=True):
         back_populates="password_reset_tokens")
 
 
+# TODO: Prevent deleting a user who is sole owner of an organization
 class User(SQLModel, table=True):
     id: Optional[int] = Field(default=None, primary_key=True)
     name: str
@@ -126,11 +134,16 @@ class User(SQLModel, table=True):
     updated_at: datetime = Field(default_factory=utc_time)
 
     organization_links: List[UserOrganizationLink] = Relationship(
-        back_populates="user"
-    )
-    organizations: List["Organization"] = Relationship(
-        back_populates="users",
-        link_model=UserOrganizationLink
+        back_populates="user",
+        sa_relationship_kwargs={
+            "cascade": "all, delete-orphan",
+            "passive_deletes": True
+        }
     )
     password_reset_tokens: List["PasswordResetToken"] = Relationship(
-        back_populates="user")
+        back_populates="user",
+        sa_relationship_kwargs={
+            "cascade": "all, delete-orphan",
+            "passive_deletes": True
+        }
+    )