learningcircuit--local-deep-research/tests/database/test_journal_rebuild_data_preservation.py

"""Data-preservation guarantees for the journals-table rebuild.

Adding ``name_lower`` and its index in the journal-quality migration
triggers a SQLite ``batch_alter_table`` rebuild (ALTER ADD COLUMN
+ index is rewritten as a full copy under the hood). This test
populates the table with 100 diverse rows *before* the migrations
touch it, runs the chain, and asserts every row survives with its
seeded columns preserved (``name``, ``quality_analysis_time``) and
``name_lower`` correctly backfilled — including diacritic, CJK, and
padded-whitespace name variants.

The ``batch_alter_table`` rebuild happens inside an Alembic
transaction, so SQLite's atomicity guarantees the table is either
fully rebuilt or untouched; a simulated mid-rebuild crash is
covered by SQLite's transaction rollback, not by our code. The
test therefore focuses on what the *output* of a successful rebuild
must look like: zero data loss, backfilled Unicode.
"""

from __future__ import annotations

import tempfile
import unicodedata
from pathlib import Path

from sqlalchemy import create_engine, inspect, text

from local_deep_research.database.alembic_runner import run_migrations
from local_deep_research.database.models import Base


def _expected_name_lower(name: str) -> str:
    """Mirror the migration's backfill expression so the test locks in
    NFKC + lower + strip semantics, not bare str.lower(). Divergence
    between writers produces silent cache misses — see
    0006_journal_quality_system.py Step 1/3.
    """
    return unicodedata.normalize("NFKC", name).lower().strip()


def _make_engine():
    tmp = Path(tempfile.mkdtemp()) / "rebuild.db"
    return create_engine(f"sqlite:///{tmp}")


def _seed(engine, n: int) -> list[tuple[str, int]]:
    """Insert ``n`` journal rows with a mix of ASCII and Unicode names."""
    rows = []
    with engine.begin() as conn:
        conn.execute(text("DELETE FROM journals"))
        for i in range(n):
            # Mix: diacritics, Asian, cased, whitespace.
            if i % 4 == 0:
                name = f"Café Research {i}"
            elif i % 4 == 1:
                name = f"JOURNAL {i}"
            elif i % 4 == 2:
                name = f"日本の学術誌 {i}"
            else:
                name = f"  Spaced Title {i}  "
            q_time = 1_700_000_000 + i
            conn.execute(
                text(
                    "INSERT INTO journals (name, quality_analysis_time) "
                    "VALUES (:n, :t)"
                ),
                {"n": name, "t": q_time},
            )
            rows.append((name, q_time))
    return rows


def test_all_rows_survive_the_chain_with_correct_backfill():
    engine = _make_engine()
    try:
        Base.metadata.create_all(engine)
        seed_rows = _seed(engine, 100)

        # Wipe any name_lower the ORM default might have set so the backfill
        # branch is the one under test.
        with engine.begin() as conn:
            conn.execute(text("UPDATE journals SET name_lower = NULL"))

        run_migrations(engine)

        # 100 rows still there; name / quality_analysis_time preserved;
        # name_lower backfilled correctly.
        with engine.begin() as conn:
            actual = conn.execute(
                text(
                    "SELECT name, name_lower, quality_analysis_time "
                    "FROM journals ORDER BY id"
                )
            ).all()

        assert len(actual) == len(seed_rows), (
            f"Row count regression: {len(actual)} vs {len(seed_rows)}"
        )
        for (seed_name, seed_t), row in zip(seed_rows, actual):
            assert row.name == seed_name
            assert row.name_lower == _expected_name_lower(seed_name)
            assert row.quality_analysis_time == seed_t
    finally:
        engine.dispose()


def test_no_orphan_tmp_table_after_migration():
    """Alembic's batch rebuild must not leave ``_alembic_tmp_journals``."""
    engine = _make_engine()
    try:
        Base.metadata.create_all(engine)
        _seed(engine, 10)
        run_migrations(engine)
        insp = inspect(engine)
        table_names = set(insp.get_table_names())
        orphans = {t for t in table_names if t.startswith("_alembic_tmp_")}
        assert not orphans, f"Orphan rebuild tables remain: {orphans}"
    finally:
        engine.dispose()