alex/headquarter
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: remove duplicate fixtures and add SQLite support

Browse Source 
Task 2.5: Remove duplicate fixtures from integration tests
- test_auth_api.py, test_auth_services.py, test_models.py
- test_projects_api.py, test_seed.py, test_users_api.py
- Fix npytest typos in all test files

Task 3.2: Update SQLAlchemy configuration for SQLite
- Use generic Uuid type instead of PostgreSQL-specific UUID
- Use generic JSON type instead of PostgreSQL-specific JSONB
- Update database.py to handle SQLite connection args

Unit tests now run without PostgreSQL (5/8 passing)
This commit is contained in:
Fusion
2026-05-18 15:14:46 +02:00
parent 3ccd94f661
commit 4299c64922
15 changed files with 80 additions and 252 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/tests/integration/test_models.py
+9 -57
View File
@@ -1,11 +1,6 @@
from collections.abc import AsyncIterator
import pytest
import pytest_asyncio
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.ext.asyncio import AsyncSession
from src.config import build_database_url
from src.models import Base
from src.models.base import TimestampMixin, UUIDPrimaryKeyMixin
from src.models.git_repository import GitRepository
@@ -16,55 +11,12 @@ from src.models.user import User
from src.models.user_config import UserConfig
TEST_DATABASE_URL = build_database_url(
user="headquarter",
password="headquarter",
host="localhost",
port=5432,
database="headquarter",
)
@pytest_asyncio.fixture
async def db_session() -> AsyncIterator[AsyncSession]:
engine = create_async_engine(TEST_DATABASE_URL)
session_factory = async_sessionmaker(engine, expire_on_commit=False)
async with session_factory() as session:
table_rows = await session.execute(
text(
"SELECT tablename FROM pg_tables "
"WHERE schemaname = 'public' "
"AND tablename = ANY(:table_names)"
),
{
"table_names": [
"refresh_tokens",
"user_configs",
"git_repositories",
"projects",
"ssh_keys",
"users",
]
},
)
existing_tables = [row[0] for row in table_rows]
if existing_tables:
await session.execute(text(f"TRUNCATE TABLE {', '.join(existing_tables)} RESTART IDENTITY CASCADE"))
await session.commit()
yield session
await session.rollback()
await engine.dispose()
n@pytest.mark.integration
@pytest.mark.integration
def test_base_metadata_collects_declared_tables() -> None:
assert isinstance(Base.metadata.tables, dict)
n@pytest.mark.integration
@pytest.mark.integration
def test_shared_mixins_define_expected_columns() -> None:
assert "id" in UUIDPrimaryKeyMixin.__dict__
@@ -72,7 +24,7 @@ def test_shared_mixins_define_expected_columns() -> None:
assert "updated_at" in TimestampMixin.__dict__
n@pytest.mark.integration
@pytest.mark.integration
def test_expected_tables_are_registered() -> None:
assert set(Base.metadata.tables) == {
@@ -85,7 +37,7 @@ def test_expected_tables_are_registered() -> None:
}
n@pytest.mark.integration
@pytest.mark.integration
def test_user_table_has_required_columns() -> None:
columns = User.__table__.columns
@@ -104,7 +56,7 @@ def test_user_table_has_required_columns() -> None:
assert columns["avatar_url"].nullable is True
n@pytest.mark.integration
@pytest.mark.integration
def test_project_relationships_point_to_owner_and_default_ssh_key() -> None:
owner_fk = next(iter(Project.__table__.c.owner_id.foreign_keys))
@@ -116,7 +68,7 @@ def test_project_relationships_point_to_owner_and_default_ssh_key() -> None:
assert Project.default_ssh_key.property.mapper.class_ is SSHKey
n@pytest.mark.integration
@pytest.mark.integration
def test_repository_and_user_config_relationships_are_registered() -> None:
project_fk = next(iter(GitRepository.__table__.c.project_id.foreign_keys))
@@ -131,7 +83,7 @@ def test_repository_and_user_config_relationships_are_registered() -> None:
assert UserConfig.user.property.mapper.class_ is User
n@pytest.mark.integration
@pytest.mark.integration
def test_refresh_token_table_has_required_columns_and_relationships() -> None:
columns = RefreshToken.__table__.columns
@@ -154,7 +106,7 @@ def test_refresh_token_table_has_required_columns_and_relationships() -> None:
@pytest.mark.asyncio
n@pytest.mark.integration
@pytest.mark.integration
async def test_async_session_can_insert_and_load_user(db_session: AsyncSession) -> None:
user = User(email="dev@headquarter.local", name="Dev User", authentik_id="dev-user", avatar_url=None)