Tests of cascade delete behaviors pass

Author: chriscarrollsmith

docs/customization.qmd

@@ -274,9 +274,9 @@ graph.write_png('static/schema.png')
 ![Database Schema](static/schema.png)
 
 
-#### Database operations
+#### Database helpers
 
-Database operations are handled by helper functions in `utils/db.py`. Key functions include:
+Database operations are facilitated by helper functions in `utils/db.py`. Key functions include:
 
 - `set_up_db()`: Initializes the database schema and default data (which we do on every application start in `main.py`)
 - `get_connection_url()`: Creates a database connection URL from environment variables in `.env`
@@ -292,3 +292,30 @@ async def get_users(session: Session = Depends(get_session)):
 ```
 
 The session automatically handles transaction management, ensuring that database operations are atomic and consistent.
+
+#### Cascade deletes
+
+Cascade deletes (in which deleting a record from one table deletes related records from another table) can be handled at either the ORM level or the database level. This template handles cascade deletes at the ORM level, via SQLModel relationships. Inside a SQLModel `Relationship`, we set:
+
+```python
+sa_relationship_kwargs={
+    "cascade": "all, delete-orphan"
+}
+```
+
+This tells SQLAlchemy to cascade all operations (e.g., `SELECT`, `INSERT`, `UPDATE`, `DELETE`) to the related table. Since this happens through the ORM, we need to be careful to do all our database operations through the ORM using supported syntax. That generally means loading database records into Python objects and then deleting those objects rather than deleting records in the database directly. 
+
+For example,
+
+```python
+session.exec(delete(Role))
+```
+
+will not trigger the cascade delete. Instead, we need to select the role objects and then delete them:
+
+```python
+for role in session.exec(select(Role)).all():
+    session.delete(role)
+```
+
+This is slower than deleting the records directly, but it makes [many-to-many relationships](https://sqlmodel.tiangolo.com/tutorial/many-to-many/create-models-with-link/#create-the-tables) much easier to manage.

tests/conftest.py

@@ -3,7 +3,7 @@
 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
+from utils.models import User, PasswordResetToken, Organization, Role
 from utils.auth import get_password_hash, create_access_token, create_refresh_token
 from main import app
 
@@ -47,15 +47,9 @@ def clean_db(session: Session):
     """
     Cleans up the database tables before each test.
     """
-    # 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)
+    for model in (PasswordResetToken, User, Role, Organization):
+        for record in session.exec(select(model)).all():
+            session.delete(record)
 
     session.commit()
 

tests/test_models.py

@@ -6,7 +6,11 @@
     RolePermissionLink,
     Organization,
     ValidPermissions,
+    User,
+    UserRoleLink,
+    PasswordResetToken,
 )
+from datetime import timedelta, datetime, UTC
 
 
 def test_permissions_persist_after_role_deletion(session: Session):
@@ -36,13 +40,8 @@ def test_permissions_persist_after_role_deletion(session: Session):
     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=delete_org_permission.id
-    )
-    role_permission_link2 = RolePermissionLink(
-        role_id=role.id, permission_id=edit_org_permission.id
-    )
-    session.add_all([role_permission_link1, role_permission_link2])
+    role.permissions.append(delete_org_permission)
+    role.permissions.append(edit_org_permission)
     session.commit()
 
     # Verify that RolePermissionLinks exist before deletion
@@ -62,3 +61,127 @@ def test_permissions_persist_after_role_deletion(session: Session):
     # Verify that RolePermissionLinks were cascade deleted
     remaining_role_permissions = session.exec(select(RolePermissionLink)).all()
     assert len(remaining_role_permissions) == 0
+
+
+def test_user_organizations_property(session: Session, test_user: User, test_organization: Organization):
+    """
+    Test that User.organizations property correctly returns all organizations
+    the user belongs to via their roles.
+    """
+    # Create a role in the test organization
+    role = Role(name="Test Role", organization_id=test_organization.id)
+    session.add(role)
+
+    # Link the user to the role
+    test_user.roles.append(role)
+    session.commit()
+
+    # Refresh the user to ensure relationships are loaded
+    session.refresh(test_user)
+
+    # Test the organizations property
+    assert len(test_user.organizations) == 1
+    assert test_user.organizations[0].id == test_organization.id
+
+
+def test_organization_users_property(session: Session, test_user: User, test_organization: Organization):
+    """
+    Test that Organization.users property correctly returns all users
+    in the organization via their roles.
+    """
+    # Create a role in the test organization
+    role = Role(name="Test Role", organization_id=test_organization.id)
+    session.add(role)
+    session.commit()
+
+    # Link the user to the role
+    test_user.roles.append(role)
+    session.commit()
+
+    # Refresh the organization to ensure relationships are loaded
+    session.refresh(test_organization)
+
+    # Test the users property
+    users_list = test_organization.users
+    assert len(users_list) == 1
+    # users_list is a list of lists due to the property implementation
+    assert test_user in users_list[0]
+
+
+def test_cascade_delete_organization(session: Session, test_user: User, test_organization: Organization):
+    """
+    Test that deleting an organization cascades properly:
+    - Deletes associated roles
+    - Deletes role-user links
+    - Does not delete users
+    """
+    # Create a role in the test organization
+    role = Role(name="Test Role", organization_id=test_organization.id)
+    session.add(role)
+    test_user.roles.append(role)
+    session.commit()
+
+    # Delete the organization
+    session.delete(test_organization)
+    session.commit()
+
+    # Verify the role was deleted
+    remaining_roles = session.exec(select(Role)).all()
+    assert len(remaining_roles) == 0
+
+    # Verify the user-role link was deleted
+    remaining_links = session.exec(select(UserRoleLink)).all()
+    assert len(remaining_links) == 0
+
+    # Verify the user still exists
+    remaining_user = session.exec(select(User)).first()
+    assert remaining_user is not None
+    assert remaining_user.id == test_user.id
+
+
+def test_password_reset_token_cascade_delete(session: Session, test_user: User):
+    """
+    Test that password reset tokens are deleted when a user is deleted
+    """
+    # Create reset tokens for the user
+    token1 = PasswordResetToken(user_id=test_user.id)
+    token2 = PasswordResetToken(user_id=test_user.id)
+    session.add(token1)
+    session.add(token2)
+    session.commit()
+
+    # Verify tokens exist
+    tokens = session.exec(select(PasswordResetToken)).all()
+    assert len(tokens) == 2
+
+    # Delete the user
+    session.delete(test_user)
+    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):
+    """
+    Test that password reset token expiration is properly set and checked
+    """
+    # Create an expired token
+    expired_token = PasswordResetToken(
+        user_id=test_user.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,
+        expires_at=datetime.now(UTC) + timedelta(hours=1)
+    )
+    session.add(valid_token)
+    session.commit()
+
+    # Verify expiration states
+    assert expired_token.is_expired()
+    assert not valid_token.is_expired()

utils/models.py

@@ -31,19 +31,14 @@ class UserRoleLink(SQLModel, table=True):
     Associates users with roles. This creates a many-to-many relationship
     between users and roles.
     """
-    id: Optional[int] = Field(default=None, primary_key=True)
-    user_id: int = Field(foreign_key="user.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_id: Optional[int] = Field(foreign_key="user.id", primary_key=True)
+    role_id: Optional[int] = Field(foreign_key="role.id", primary_key=True)
 
 
 class RolePermissionLink(SQLModel, table=True):
-    id: Optional[int] = Field(default=None, primary_key=True)
-    role_id: int = Field(foreign_key="role.id")
-    permission_id: int = Field(foreign_key="permission.id")
-    created_at: datetime = Field(default_factory=utc_time)
-    updated_at: datetime = Field(default_factory=utc_time)
+    role_id: Optional[int] = Field(foreign_key="role.id", primary_key=True)
+    permission_id: Optional[int] = Field(
+        foreign_key="permission.id", primary_key=True)
 
 
 class Permission(SQLModel, table=True):
@@ -72,8 +67,7 @@ class Organization(SQLModel, table=True):
     roles: List["Role"] = Relationship(
         back_populates="organization",
         sa_relationship_kwargs={
-            "cascade": "all, delete-orphan",
-            "passive_deletes": True
+            "cascade": "all, delete-orphan"
         }
     )
 
@@ -98,7 +92,8 @@ class Role(SQLModel, table=True):
     """
     id: Optional[int] = Field(default=None, primary_key=True)
     name: str
-    organization_id: int = Field(foreign_key="organization.id")
+    organization_id: int = Field(
+        foreign_key="organization.id")
     created_at: datetime = Field(default_factory=utc_time)
     updated_at: datetime = Field(default_factory=utc_time)
 
@@ -125,6 +120,12 @@ class PasswordResetToken(SQLModel, table=True):
     user: Optional["User"] = Relationship(
         back_populates="password_reset_tokens")
 
+    def is_expired(self) -> bool:
+        """
+        Check if the token has expired
+        """
+        return datetime.now(UTC) > self.expires_at.replace(tzinfo=UTC)
+
 
 # TODO: Prevent deleting a user who is sole owner of an organization
 class User(SQLModel, table=True):
@@ -143,8 +144,7 @@ class User(SQLModel, table=True):
     password_reset_tokens: List["PasswordResetToken"] = Relationship(
         back_populates="user",
         sa_relationship_kwargs={
-            "cascade": "all, delete-orphan",
-            "passive_deletes": True
+            "cascade": "all, delete-orphan"
         }
     )