|
1 |
| -from sqlmodel import Session, select |
| 1 | +import warnings |
| 2 | +from sqlmodel import Session, select, inspect |
2 | 3 | from utils.db import (
|
3 | 4 | get_connection_url,
|
4 | 5 | assign_permissions_to_role,
|
5 | 6 | create_default_roles,
|
6 | 7 | create_permissions,
|
| 8 | + tear_down_db, |
| 9 | + set_up_db, |
7 | 10 | )
|
8 | 11 | from utils.models import Role, Permission, Organization, RolePermissionLink, ValidPermissions
|
| 12 | +from sqlalchemy import create_engine |
9 | 13 |
|
10 | 14 |
|
11 | 15 | def test_get_connection_url():
|
@@ -120,3 +124,66 @@ def test_assign_permissions_to_role_duplicate_check(session: Session, test_organ
|
120 | 124 | )
|
121 | 125 | ).all()
|
122 | 126 | assert len(link_count) == 1
|
| 127 | + |
| 128 | + |
| 129 | +def test_set_up_db_creates_tables(): |
| 130 | + """Test that set_up_db creates all expected tables without warnings""" |
| 131 | + # First tear down any existing tables |
| 132 | + tear_down_db() |
| 133 | + |
| 134 | + # Run set_up_db with drop=False since we just cleaned up |
| 135 | + set_up_db(drop=False) |
| 136 | + |
| 137 | + # Use SQLAlchemy inspect to check tables |
| 138 | + engine = create_engine(get_connection_url()) |
| 139 | + inspector = inspect(engine) |
| 140 | + table_names = inspector.get_table_names() |
| 141 | + |
| 142 | + # Check for core tables |
| 143 | + expected_tables = { |
| 144 | + "user", |
| 145 | + "organization", |
| 146 | + "role", |
| 147 | + "permission", |
| 148 | + "role_permission_link", |
| 149 | + "password_reset_token" |
| 150 | + } |
| 151 | + assert expected_tables.issubset(set(table_names)) |
| 152 | + |
| 153 | + # Verify permissions were created |
| 154 | + with Session(engine) as session: |
| 155 | + permissions = session.exec(select(Permission)).all() |
| 156 | + assert len(permissions) == len(ValidPermissions) |
| 157 | + |
| 158 | + # Clean up |
| 159 | + tear_down_db() |
| 160 | + engine.dispose() |
| 161 | + |
| 162 | + |
| 163 | +def test_set_up_db_drop_flag(): |
| 164 | + """Test that set_up_db's drop flag properly recreates tables""" |
| 165 | + # Set up db with drop=True |
| 166 | + engine = create_engine(get_connection_url()) |
| 167 | + set_up_db(drop=True) |
| 168 | + |
| 169 | + # Create a new session for this test |
| 170 | + with Session(engine) as session: |
| 171 | + # Verify valid permissions exist |
| 172 | + permissions = session.exec(select(Permission)).all() |
| 173 | + assert len(permissions) == len(ValidPermissions) |
| 174 | + |
| 175 | + # Create an organization |
| 176 | + org = Organization(name="Test Organization") |
| 177 | + session.add(org) |
| 178 | + session.commit() |
| 179 | + |
| 180 | + # Set up db with drop=False |
| 181 | + set_up_db(drop=False) |
| 182 | + |
| 183 | + # Verify organization exists |
| 184 | + assert session.exec(select(Organization).where( |
| 185 | + Organization.name == "Test Organization")).first() is not None |
| 186 | + |
| 187 | + # Clean up |
| 188 | + tear_down_db() |
| 189 | + engine.dispose() |
0 commit comments