LLeMbas/tests/test_migrations.py

"""The upgrade path, which every existing install takes to reach 1.0.0.

There is no Alembic here by design -- hard rule 4, additive-only, with
`db/migrations.py:sync_schema` deriving the change from the models by diffing
them against the live database. That makes the differ the *only* thing standing
between an 0.8.x deployment and a broken one, and until this file it had no test
that exercised it as a migration at all.

`tests/conftest.py` runs `create_all` and *then* `sync_schema`, so the schema it
is handed is always already current: the differ finds nothing missing, does
nothing, and reports success. Every run of the suite proved that a no-op is a
no-op. The two unit tests that did exist covered `_add_column_sql`'s DDL string
and never touched a database.

So these build an **old-shaped database with rows in it** and upgrade it for
real. The shape is not invented: `OLD_TABLES` and `OLD_COLUMNS` are what
actually arrived between `0.8.1` and this release, taken from
`git diff 00ce04a..HEAD -- src/lembas/db/models/`. `test_the_recorded_shape_is_still_real`
is what stops that list rotting into a test that upgrades nothing.
"""

from __future__ import annotations

import pytest
from sqlalchemy import inspect, text

from lembas.db.base import Base
from lembas.db.migrations import ensure_fts, sync_schema
from lembas.db.session import get_engine

# Tables that did not exist at 0.8.1. `sync_schema` has to create them.
OLD_TABLES = ("chunks", "push_subscriptions", "usage")

# Columns added to tables that already existed, and therefore already had rows.
# These are the interesting half: a new *table* is empty by definition, but a
# new *column* has to arrive beside data somebody cares about.
OLD_COLUMNS = (
    ("ssh_profiles", "resolves_here"),
    ("chats", "parent_chat_id"),
    ("chats", "unattended"),
    ("reports", "unread_notified"),
    ("groups", "limits_json"),
)


def _rollback_to_0_8_1(engine) -> None:
    """Take a current database back to the shape 0.8.1 left behind.

    Backwards rather than forwards because the alternative is importing two
    versions of the models into one interpreter, which is not a thing Python
    will do -- `lembas.db.models` is already imported and registered on one
    `Base.metadata`. Dropping what was added produces the same *shape* the
    differ has to repair, which is what is under test.
    """
    with engine.begin() as connection:
        for table in OLD_TABLES:
            connection.execute(text(f"DROP TABLE IF EXISTS {table}"))
        for table, column in OLD_COLUMNS:
            connection.execute(text(f"ALTER TABLE {table} DROP COLUMN {column}"))


def _columns(engine, table: str) -> set[str]:
    return {column["name"] for column in inspect(engine).get_columns(table)}


def test_the_recorded_shape_is_still_real():
    """The rollback above names tables and columns by hand, and a name that has
    since been removed or renamed would make it silently upgrade nothing --
    a test that passes because it tested an empty set.
    """
    tables = Base.metadata.tables

    for table in OLD_TABLES:
        assert table in tables, f"{table} is no longer a table; fix OLD_TABLES"
    for table, column in OLD_COLUMNS:
        assert table in tables, table
        assert column in tables[table].c, f"{table}.{column} is gone; fix OLD_COLUMNS"


def test_an_0_8_1_database_with_data_upgrades(db):
    """The whole point. Rows written before the upgrade must still be there
    afterwards, with their values, and the new columns must exist beside them.

    A failure here is somebody's instance not starting after pressing Update,
    or -- worse and quieter -- starting with a column that silently reads NULL.
    """
    engine = get_engine()

    # Data first, while the schema still has the columns the ORM expects.
    from lembas.db.models import Chat, Group, User

    owner = User(name="Frodo", email="f@shire.test", password_hash="x")  # noqa: S106
    db.add(owner)
    db.commit()
    group = Group(name="Fellowship")
    db.add(group)
    db.commit()
    chat = Chat(user_id=owner.id, model_id="mithril", title="A chat from before")
    db.add(chat)
    db.commit()
    chat_id, owner_id, group_id = chat.id, owner.id, group.id
    db.close()

    _rollback_to_0_8_1(engine)

    # Confirm the rollback actually removed things, or the assertions below
    # would pass against a database that was never old.
    assert "chunks" not in inspect(engine).get_table_names()
    assert "unattended" not in _columns(engine, "chats")

    changes = sync_schema(engine)

    assert changes, "the differ reported nothing to do on an old database"
    for table in OLD_TABLES:
        assert table in inspect(engine).get_table_names(), table
    for table, column in OLD_COLUMNS:
        assert column in _columns(engine, table), f"{table}.{column}"

    # And the rows are still what they were.
    with engine.connect() as connection:
        row = connection.execute(
            text("SELECT title, model_id, user_id FROM chats WHERE id = :id"),
            {"id": chat_id},
        ).one()
        assert row.title == "A chat from before"
        assert row.model_id == "mithril"
        assert row.user_id == owner_id
        assert (
            connection.execute(
                text("SELECT name FROM groups WHERE id = :id"), {"id": group_id}
            ).scalar()
            == "Fellowship"
        )


def test_a_nullable_column_arrives_empty_rather_than_defaulted(db):
    """A nullable column is added with **no** default, so an existing row reads
    NULL -- the value the model treats as absent.

    An earlier version defaulted every column by type, which meant an added
    foreign key arrived as `""` on old rows, and every "is this set?" check
    downstream was wrong about them. `parent_chat_id` is exactly that shape: a
    chat that predates subagents is not a helper, and `""` would not say so.
    """
    engine = get_engine()

    from lembas.db.models import Chat, User

    owner = User(name="Sam", email="s@shire.test", password_hash="x")  # noqa: S106
    db.add(owner)
    db.commit()
    chat = Chat(user_id=owner.id, model_id="m")
    db.add(chat)
    db.commit()
    chat_id = chat.id
    db.close()

    _rollback_to_0_8_1(engine)
    sync_schema(engine)

    with engine.connect() as connection:
        parent = connection.execute(
            text("SELECT parent_chat_id FROM chats WHERE id = :id"), {"id": chat_id}
        ).scalar()
    assert parent is None, "an added foreign key must be absent, not empty"


def test_a_not_null_column_backfills_every_existing_row(db):
    """SQLite refuses to add a NOT NULL column without a default, so one is
    derived from the type. `chats.unattended` is a boolean: an old row has to
    come back False rather than NULL, or every `if chat.unattended` downstream
    reads a null as a value.
    """
    engine = get_engine()

    from lembas.db.models import Chat, User

    owner = User(name="Merry", email="m@shire.test", password_hash="x")  # noqa: S106
    db.add(owner)
    db.commit()
    db.add(Chat(user_id=owner.id, model_id="m"))
    db.commit()
    db.close()

    _rollback_to_0_8_1(engine)
    sync_schema(engine)

    with engine.connect() as connection:
        values = connection.execute(text("SELECT unattended FROM chats")).scalars().all()
    assert values, "no rows survived the upgrade"
    assert all(value in (0, False) for value in values), values


def test_upgrading_twice_changes_nothing_the_second_time(db):
    """It runs on every startup and has to converge. A second pass reporting
    work would mean it was re-adding something, which on SQLite is an error that
    stops the application booting."""
    engine = get_engine()
    _rollback_to_0_8_1(engine)

    assert sync_schema(engine)
    assert sync_schema(engine) == []


def test_the_search_indexes_are_rebuilt_when_missing(db):
    """FTS5 tables are not SQLAlchemy models, so `create_all` cannot make them
    and the column differ cannot see them. `ensure_fts` is what converges them,
    and it runs at every startup for the same reason the column sync does --
    an instance whose indexes were dropped must repair itself rather than
    return nothing and call it an empty library.
    """
    engine = get_engine()
    with engine.begin() as connection:
        connection.execute(text("DROP TABLE IF EXISTS documents_fts"))

    assert "documents_fts" in ensure_fts(engine)
    assert "documents_fts" in inspect(engine).get_table_names()
    # Converges: a second call has nothing left to make.
    assert ensure_fts(engine) == []


@pytest.mark.parametrize("table", OLD_TABLES)
def test_each_new_table_is_usable_after_the_upgrade(db, table):
    """Created is not the same as correct. A table the differ made but got the
    columns wrong on fails at the first insert, which is the first *reply* on an
    upgraded instance rather than at startup."""
    engine = get_engine()
    _rollback_to_0_8_1(engine)
    sync_schema(engine)

    declared = {column.name for column in Base.metadata.tables[table].c}
    assert _columns(engine, table) == declared