ss-tools/backend/tests/integration/test_alembic_user_id_migration.py

# #region Test.Alembic.TaskRecordsUserId [C:3] [TYPE Module] [SEMANTICS test,alembic,migration,task,postgres]
# @BRIEF Verifies the baseline schema includes task_records.user_id.
# @RELATION BINDS_TO -> [EXT:Alembic:MigrationChain]
# @TEST_FIXTURE: migration_revisions -> INLINE_JSON
# @TEST_EDGE: missing_field -> Baseline schema must expose task_records.user_id.
# @TEST_EDGE: invalid_type -> Baseline schema is created by Alembic, not ORM startup.
# @TEST_INVARIANT: task_records_user_id_baseline -> VERIFIED_BY: [Test.Alembic.TestUserIdMigrationAddsColumn]
# @RATIONALE SQLite and metadata.create_all() cannot expose an unapplied PostgreSQL migration;
#            an isolated Testcontainers database exercises Alembic's actual DDL path.
# @REJECTED Stamping the revision or applying raw ALTER TABLE was rejected because each can
#           conceal a migration chain failure while falsely marking the schema current.

import os
from pathlib import Path
import pytest

from sqlalchemy import create_engine, inspect, text


# #region Test.Alembic.MigrationDatabaseUrl [C:2] [TYPE Function]
# @BRIEF Provide a pristine isolated database created via db_factory in the shared PG container.
# @POST Returns a host-accessible connection URL for an empty database.
# @RATIONALE Uses db_factory from the shared Postgres container instead of starting
#   a separate PostgreSQL container. This is faster and uses fewer Docker resources.
#   The database is created as a fresh, empty database — no tables exist, no Alembic
#   has been run. This preserves the pristine-schema semantics required for testing
#   Alembic migrations from a clean state.
# @REJECTED Starting a separate PostgresContainer was rejected — it adds ~5s per test
#   and unnecessary Docker overhead when the shared container is already running.
@pytest.fixture
def migration_database_url(db_factory) -> str:
    """Provide a pristine PostgreSQL database from the shared container via db_factory.

    Returns a host-accessible Postgres URL. The database is empty (no tables).
    """
    result = db_factory["create_db"]("_alembic_test")
    return result["host_url"]


# #endregion Test.Alembic.MigrationDatabaseUrl


# #region Test.Alembic.TestUserIdMigrationAddsColumn [C:2] [TYPE Function] [SEMANTICS test,alembic,migration,task,postgres]
# @BRIEF Upgrades a pristine PostgreSQL schema through the current baseline.
def test_user_id_migration_adds_column(monkeypatch: pytest.MonkeyPatch, migration_database_url: str) -> None:
    """Alembic baseline creates task_records.user_id without runtime schema creation."""
    monkeypatch.setenv("DATABASE_URL", migration_database_url)
    _run_alembic_upgrade("head")

    engine = create_engine(migration_database_url)
    try:
        columns = {column["name"] for column in inspect(engine).get_columns("task_records")}
        assert "user_id" in columns

        _run_alembic_upgrade("head")
        assert "user_id" in {column["name"] for column in inspect(engine).get_columns("task_records")}
    finally:
        engine.dispose()
# #endregion Test.Alembic.TestUserIdMigrationAddsColumn


def test_prepare_database_resets_any_legacy_revision(
    monkeypatch: pytest.MonkeyPatch,
    migration_database_url: str,
) -> None:
    """An explicit reset replaces any old Alembic chain with the current baseline."""
    engine = create_engine(migration_database_url)
    try:
        with engine.begin() as connection:
            connection.execute(text("CREATE TABLE alembic_version (version_num VARCHAR(64) NOT NULL)"))
            connection.execute(
                text("INSERT INTO alembic_version (version_num) VALUES ('y4z5a6b7c8d9')")
            )
            connection.execute(text("CREATE TABLE legacy_probe (id INTEGER PRIMARY KEY)"))

        monkeypatch.setenv("DATABASE_URL", migration_database_url)
        # Orphaned revisions are reset independently of stale Compose defaults.
        monkeypatch.setenv("RESET_DATABASE_SCHEMA", "false")
        monkeypatch.delenv("RESET_DATABASE_FROM_REVISION", raising=False)
        monkeypatch.delenv("RESET_DATABASE_NAME", raising=False)

        from src.scripts.prepare_database import prepare_database

        prepare_database()

        tables = set(inspect(engine).get_table_names())
        assert "legacy_probe" not in tables
        assert "task_records" in tables
        with engine.connect() as connection:
            revision = connection.execute(text("SELECT version_num FROM alembic_version")).scalar_one()
        assert revision == "0001_baseline"
    finally:
        engine.dispose()


def test_prepare_database_refuses_unversioned_nonempty_schema(
    monkeypatch: pytest.MonkeyPatch,
    migration_database_url: str,
) -> None:
    """The universal reset must not erase an unrelated unversioned database."""
    engine = create_engine(migration_database_url)
    try:
        with engine.begin() as connection:
            connection.execute(text("CREATE TABLE unrelated_data (id INTEGER PRIMARY KEY)"))

        monkeypatch.setenv("DATABASE_URL", migration_database_url)
        monkeypatch.setenv("RESET_DATABASE_SCHEMA", "true")

        from src.scripts.prepare_database import prepare_database

        with pytest.raises(RuntimeError, match="non-empty but has no Alembic revision"):
            prepare_database()
        assert "unrelated_data" in inspect(engine).get_table_names()
    finally:
        engine.dispose()


# #region _run_alembic_upgrade [C:1] [TYPE Function]
def _run_alembic_upgrade(revision: str) -> None:
    """Run a named revision against the DATABASE_URL injected for this test."""
    from alembic.config import Config

    from alembic import command

    backend_dir = Path(__file__).parents[2]
    config = Config(str(backend_dir / "alembic.ini"))
    config.set_main_option("script_location", str(backend_dir / "alembic"))
    config.set_main_option("sqlalchemy.url", os.environ["DATABASE_URL"])
    command.upgrade(config, revision)
# #endregion _run_alembic_upgrade


# #endregion Test.Alembic.TaskRecordsUserId