fix: legacy DB support + QA audit fixes
- Entrypoint: for legacy DBs, add is_admin via raw SQL then stamp head (instead of running full migration chain which fails on existing tables) - Fixed admin_api_keys.py Pydantic config (dict → ConfigDict) - Added tests/test_alembic_migrations.py (PostgreSQL-only integration tests) - Fixed infinite recursion in _create_table_if_not_exists (QA catch)
This commit is contained in:
@@ -11,7 +11,7 @@
|
||||
from datetime import datetime
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
from pydantic import BaseModel, Field
|
||||
from pydantic import BaseModel, ConfigDict, Field
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from ...core.auth.api_key import generate_api_key
|
||||
@@ -62,7 +62,7 @@ class ApiKeyListItem(BaseModel):
|
||||
expires_at: datetime | None
|
||||
last_used_at: datetime | None
|
||||
|
||||
model_config = {"from_attributes": True}
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
# #endregion ApiKeyListItem
|
||||
|
||||
|
||||
|
||||
126
backend/tests/test_alembic_migrations.py
Normal file
126
backend/tests/test_alembic_migrations.py
Normal file
@@ -0,0 +1,126 @@
|
||||
# #region TestAlembicMigrations [C:3] [TYPE Module] [SEMANTICS tests, alembic, migration, schema, idempotent]
|
||||
# @BRIEF Verifies Alembic migration chain: fresh upgrade and legacy stamp.
|
||||
# @RELATION BINDS_TO -> [ed310b33f02c]
|
||||
# @INVARIANT Running `alembic upgrade head` twice is idempotent (no-op on second run) on PostgreSQL.
|
||||
# @INVARIANT Legacy database with existing tables can be stamped after adding boot-critical columns.
|
||||
# @PRE Alembic is installed, migration files exist in alembic/versions/.
|
||||
# @POST Schema matches expected state after upgrade.
|
||||
# @TEST_INVARIANT: alembic_fresh_upgrade -> VERIFIED_BY: [test_fresh_upgrade_creates_tables]
|
||||
# @TEST_INVARIANT: alembic_stamp_legacy -> VERIFIED_BY: [test_legacy_database_stamp]
|
||||
# @REJECTED Full migration chain idempotency on SQLite — initial migration uses
|
||||
# op.create_index() which conflicts with SQLite auto-created indexes
|
||||
# for UNIQUE constraints. Migrations are designed for PostgreSQL.
|
||||
# Legacy databases use stamp + selective raw SQL in entrypoint instead.
|
||||
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
# Ensure backend/src is importable for model metadata
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
||||
|
||||
# Alembic migrations use PostgreSQL-specific features (CREATE INDEX may conflict
|
||||
# with SQLite auto-indexes for UNIQUE constraints). Skip full-chain tests on SQLite.
|
||||
_DB_BACKEND = os.environ.get("DATABASE_URL", "sqlite://").split(":")[0]
|
||||
_REQUIRES_PG = pytest.mark.skipif(
|
||||
_DB_BACKEND != "postgresql",
|
||||
reason="Alembic migrations are PostgreSQL-specific; use postgresql:// DATABASE_URL to test",
|
||||
)
|
||||
|
||||
|
||||
# #region test_fresh_upgrade_creates_tables [C:3] [TYPE Function]
|
||||
# @BRIEF Run alembic upgrade head on a fresh PostgreSQL DB and verify key tables exist.
|
||||
# @POST All expected tables exist after upgrade.
|
||||
# @TEST_INVARIANT: alembic_fresh_upgrade -> VERIFIED_BY: [test_fresh_upgrade_creates_tables]
|
||||
@_REQUIRES_PG
|
||||
def test_fresh_upgrade_creates_tables() -> None:
|
||||
"""Fresh Alembic upgrade creates all expected tables (PostgreSQL only)."""
|
||||
from sqlalchemy import create_engine, inspect
|
||||
from sqlalchemy import text as sa_text
|
||||
backend_dir = Path(__file__).parent.parent
|
||||
_run_alembic_upgrade()
|
||||
engine = create_engine(os.environ["DATABASE_URL"])
|
||||
inspector = inspect(engine)
|
||||
tables = set(inspector.get_table_names())
|
||||
expected_tables = {
|
||||
"environments",
|
||||
"translation_jobs",
|
||||
"users",
|
||||
"roles",
|
||||
"permissions",
|
||||
"task_records",
|
||||
"alembic_version",
|
||||
}
|
||||
missing = expected_tables - tables
|
||||
assert not missing, f"Missing tables after Alembic upgrade: {missing}"
|
||||
with engine.connect() as conn:
|
||||
result = conn.execute(sa_text("SELECT version_num FROM alembic_version"))
|
||||
version = result.scalar_one()
|
||||
assert version is not None and len(version) > 0
|
||||
roles_cols = {c["name"] for c in inspector.get_columns("roles")}
|
||||
assert "is_admin" in roles_cols, "roles.is_admin column missing after Alembic upgrade"
|
||||
engine.dispose()
|
||||
# #endregion test_fresh_upgrade_creates_tables
|
||||
|
||||
|
||||
# #region test_legacy_database_stamp [C:2] [TYPE Function]
|
||||
# @BRIEF Pre-populate tables via create_all(), then verify entrypoint-style stamp works.
|
||||
# @POST alembic_version created, is_admin column present after stamp + raw SQL.
|
||||
@_REQUIRES_PG
|
||||
def test_legacy_database_stamp() -> None:
|
||||
"""Alembic stamp works on a database that already has tables (PostgreSQL only)."""
|
||||
from sqlalchemy import create_engine, inspect
|
||||
from sqlalchemy import text as sa_text
|
||||
from src.models.mapping import Base
|
||||
from src.models import auth as _auth # noqa: F401
|
||||
from src.models import task as _task # noqa: F401
|
||||
from alembic.config import Config as AlembicConfig
|
||||
from alembic import command as alembic_command
|
||||
|
||||
engine = create_engine(os.environ["DATABASE_URL"])
|
||||
# Drop alembic_version if exists to simulate legacy state
|
||||
with engine.begin() as conn:
|
||||
conn.execute(sa_text("DROP TABLE IF EXISTS alembic_version CASCADE"))
|
||||
engine.dispose()
|
||||
# Tables already exist from previous runs (create_all)
|
||||
# Add is_admin column (simulating entrypoint raw SQL)
|
||||
engine = create_engine(os.environ["DATABASE_URL"])
|
||||
with engine.begin() as conn:
|
||||
inspector = inspect(conn)
|
||||
roles_cols = {c["name"] for c in inspector.get_columns("roles")}
|
||||
if "is_admin" not in roles_cols:
|
||||
conn.execute(
|
||||
sa_text("ALTER TABLE roles ADD COLUMN is_admin BOOLEAN NOT NULL DEFAULT FALSE")
|
||||
)
|
||||
conn.execute(sa_text("UPDATE roles SET is_admin = TRUE WHERE name = 'Admin'"))
|
||||
engine.dispose()
|
||||
# Stamp head (entrypoint behavior for legacy DB)
|
||||
backend_dir = Path(__file__).parent.parent
|
||||
alembic_cfg = AlembicConfig(str(backend_dir / "alembic.ini"))
|
||||
alembic_cfg.set_main_option("script_location", str(backend_dir / "alembic"))
|
||||
alembic_command.stamp(alembic_cfg, "head")
|
||||
# Verify
|
||||
engine = create_engine(os.environ["DATABASE_URL"])
|
||||
inspector = inspect(engine)
|
||||
tables = set(inspector.get_table_names())
|
||||
assert "alembic_version" in tables, "alembic_version table not created by stamp"
|
||||
roles_cols = {c["name"] for c in inspector.get_columns("roles")}
|
||||
assert "is_admin" in roles_cols
|
||||
engine.dispose()
|
||||
# #endregion test_legacy_database_stamp
|
||||
|
||||
|
||||
# #region _run_alembic_upgrade [C:1] [TYPE Function]
|
||||
def _run_alembic_upgrade() -> None:
|
||||
"""Run alembic upgrade head programmatically."""
|
||||
from alembic.config import Config as AlembicConfig
|
||||
from alembic import command as alembic_command
|
||||
backend_dir = Path(__file__).parent.parent
|
||||
alembic_cfg = AlembicConfig(str(backend_dir / "alembic.ini"))
|
||||
alembic_cfg.set_main_option("script_location", str(backend_dir / "alembic"))
|
||||
alembic_command.upgrade(alembic_cfg, "head")
|
||||
# #endregion _run_alembic_upgrade
|
||||
|
||||
# #endregion TestAlembicMigrations
|
||||
@@ -175,9 +175,25 @@ elif [ $_db_state -eq 1 ]; then
|
||||
alembic upgrade head 2>&1 | sed 's/^/[entrypoint] /'
|
||||
echo "[entrypoint] ✅ Alembic initial migration applied"
|
||||
else
|
||||
echo "[entrypoint] Legacy database — stamping Alembic head (tables already exist)..."
|
||||
echo "[entrypoint] Legacy database — adding critical migration columns..."
|
||||
# Add boot-critical columns that Alembic migrations would add
|
||||
python3 -c "
|
||||
import os
|
||||
from sqlalchemy import create_engine, inspect, text
|
||||
e = create_engine(os.environ['DATABASE_URL'])
|
||||
with e.begin() as conn:
|
||||
inspector = inspect(conn)
|
||||
roles_cols = {c['name'] for c in inspector.get_columns('roles')}
|
||||
if 'is_admin' not in roles_cols:
|
||||
conn.execute(text('ALTER TABLE roles ADD COLUMN is_admin BOOLEAN NOT NULL DEFAULT FALSE'))
|
||||
conn.execute(text(\"UPDATE roles SET is_admin = TRUE WHERE name = 'Admin'\"))
|
||||
print(' roles.is_admin column added')
|
||||
else:
|
||||
print(' roles.is_admin already exists')
|
||||
"
|
||||
echo "[entrypoint] Stamping Alembic head (tables already exist)..."
|
||||
alembic stamp head 2>&1 | sed 's/^/[entrypoint] /'
|
||||
echo "[entrypoint] ✅ Alembic stamped head, schema assumed up to date"
|
||||
echo "[entrypoint] ✅ Alembic stamped head, schema ready"
|
||||
fi
|
||||
|
||||
bootstrap_admin
|
||||
|
||||
Reference in New Issue
Block a user