shopdb-flask/tests/test_plugin_migrations.py

"""Per-plugin Alembic migration-chain guard tests.

Ownership cutover (ADR-008): the core Alembic chain created every table that
exists through its head (`7d16_directoryemployees`), including the plugin
tables. From that point on, each bundled plugin that owns tables carries its
own chain under `plugins/<name>/migrations/`. The `0001` migration in each
chain is a stamp-only no-op anchor: the core chain already built the tables, so
there is nothing to create; the anchor just gives the plugin chain a base that
`flask plugin upgrade-all` stamps into `alembic_version_<plugin>`.

These tests pin that contract:
  * PLUGIN_TABLE_OWNERS stays in sync with what the models declare.
  * Every plugin that owns tables has a valid single-head chain.
  * The anchor migrations are genuine no-ops.
  * `flask plugin upgrade-all` runs clean on a fresh DB and is idempotent.
"""
import ast
from pathlib import Path

import pytest

from shopdb.plugins.alembic_template import (
    PLUGIN_TABLE_OWNERS,
    _get_plugin_metadata,
)

PLUGINS_DIR = Path(__file__).resolve().parent.parent / 'plugins'

# Plugins that own tables carry a migration chain; blueprint-only plugins do
# not. Today every bundled plugin owns tables, so this is the full set.
TABLE_OWNING_PLUGINS = tuple(sorted(PLUGIN_TABLE_OWNERS))

# The ADR-008 cutover froze this exact set of ten plugins whose tables the core
# chain had already created. Their 0001 revision is a stamp-only no-op anchor.
# Plugins built AFTER the cutover (e.g. measuringtools) are NOT in this list:
# their 0001 is a real baseline that genuinely creates their tables, so the
# no-op assertion must not apply to them. This is a frozen list on purpose - a
# newly discovered plugin does not silently get treated as a cutover no-op.
CUTOVER_PLUGINS = (
    'computers', 'employees', 'knowledgebase', 'machines', 'network',
    'notifications', 'printers', 'slides', 'usb', 'warranty',
)

# Expected head revision id per table-owning plugin, so the upgrade-all test can
# check both the cutover anchors and post-cutover baselines. Cutover plugins
# stamp '<plugin>0001anchor'; measuringtools stamps its real baseline id.
EXPECTED_HEAD_REVISION = {plugin: f'{plugin}0001anchor' for plugin in CUTOVER_PLUGINS}
EXPECTED_HEAD_REVISION['measuringtools'] = 'measuringtools0001baseline'
# backups is also post-cutover: its 0001 really creates backuprevisions, and
# 0002 adds lastseenat - when a config was last CONFIRMED unchanged, which dedup
# otherwise throws away.
EXPECTED_HEAD_REVISION['backups'] = 'backups0003clearlastseen'
# geenforce adds the content-addressed blob store (manifestblobs) on top of its
# baseline.
EXPECTED_HEAD_REVISION['geenforce'] = 'geenforce0004minlib'
# printers adds the printersupplyalerts crossing-state table on top of its
# anchor, then the exact INF driver name Add-PrinterDriver needs, a vendor on
# the driver, and finally printerobservedqueues - what a bay reported it HAS,
# kept apart from what it was assigned.
EXPECTED_HEAD_REVISION['printers'] = 'printers0005observedqueues'
# machines (renamed from equipment) keeps its original anchor id and adds the
# rename revision on top, so its head is not the f-string default.
EXPECTED_HEAD_REVISION['machines'] = 'machines0002rename'
# employees adds the photofilename column on top of its cutover anchor.
EXPECTED_HEAD_REVISION['employees'] = 'employees0002photo'
# usb drops the dead usbcheckouts.machineid column on top of its anchor.
EXPECTED_HEAD_REVISION['usb'] = 'usb0002dropmachineid'
# network links a device to a catalog model, which is where its photo comes
# from - machines, PCs and printers already had that link.
EXPECTED_HEAD_REVISION['network'] = 'network0003prefix'
# printedparts is post-cutover: its 0001 really creates its tables; 0004 adds
# the per-transaction revision column.
EXPECTED_HEAD_REVISION['printedparts'] = 'printedparts0004txnrev'
# Built after the ADR-008 cutover, so its 0001 genuinely creates the table
# rather than stamping an anchor. Tech Tools owns a table now because secure
# send needs one; every other tool in that plugin is browser-side.
EXPECTED_HEAD_REVISION['tools'] = 'tools0001securesend'
# notifications indexes businessunitid, then adds the per-type grace window and
# the shared board category.
EXPECTED_HEAD_REVISION['notifications'] = 'notifications0005boardorder'
# warranty adds the proof-of-cover document columns on top of its anchor.
EXPECTED_HEAD_REVISION['warranty'] = 'warranty0002proof'

# Plugins built after the cutover: their 0001 baseline really creates tables the
# core chain never owned.
POST_CUTOVER_PLUGINS = tuple(p for p in PLUGIN_TABLE_OWNERS if p not in CUTOVER_PLUGINS)


def _declared_tablenames(plugin: str) -> set:
    """Scan a plugin's models package for every __tablename__ literal.

    Static parse (no import) so the test can compare what the code declares
    against PLUGIN_TABLE_OWNERS without side effects.
    """
    names = set()
    models_dir = PLUGINS_DIR / plugin / 'models'
    if not models_dir.exists():
        return names
    for source in models_dir.glob('*.py'):
        tree = ast.parse(source.read_text())
        for node in ast.walk(tree):
            if isinstance(node, ast.Assign):
                targets = [t.id for t in node.targets if isinstance(t, ast.Name)]
                if '__tablename__' in targets and isinstance(node.value, ast.Constant):
                    names.add(node.value.value)
    return names


def test_table_owners_match_declared_models():
    """Every plugin that declares a __tablename__ is registered in
    PLUGIN_TABLE_OWNERS, and vice versa. Catches a new plugin table that
    forgot to update the ownership map."""
    declared = {p.name for p in PLUGINS_DIR.iterdir()
                if p.is_dir() and _declared_tablenames(p.name)}
    assert declared == set(PLUGIN_TABLE_OWNERS), (
        f"PLUGIN_TABLE_OWNERS keys {set(PLUGIN_TABLE_OWNERS)} do not match "
        f"plugins declaring tables {declared}"
    )


@pytest.mark.parametrize('plugin', TABLE_OWNING_PLUGINS)
def test_bundled_plugin_has_table_owner_entry(plugin):
    """Every table-owning plugin appears in PLUGIN_TABLE_OWNERS with at least
    one table, documenting which tables it contributes to the schema."""
    assert plugin in PLUGIN_TABLE_OWNERS
    assert len(PLUGIN_TABLE_OWNERS[plugin]) > 0


@pytest.mark.parametrize('plugin', TABLE_OWNING_PLUGINS)
def test_owned_tables_match_declared_models(plugin):
    """The tables named in PLUGIN_TABLE_OWNERS are exactly the ones the
    plugin's models declare. Catches drift in either direction."""
    assert set(PLUGIN_TABLE_OWNERS[plugin]) == _declared_tablenames(plugin)


@pytest.mark.parametrize('plugin', TABLE_OWNING_PLUGINS)
def test_plugin_metadata_has_all_owned_tables(plugin, app):
    """The MetaData filtered to a plugin's owned tables actually contains
    every table named in PLUGIN_TABLE_OWNERS."""
    with app.app_context():
        md = _get_plugin_metadata(plugin)
        owned = set(PLUGIN_TABLE_OWNERS[plugin])
        missing = owned - set(md.tables.keys())
        assert not missing, f"Plugin {plugin}: owned tables not in metadata: {missing}"


@pytest.mark.parametrize('plugin', TABLE_OWNING_PLUGINS)
def test_plugin_has_migration_chain(plugin):
    """Every table-owning plugin has a migrations dir with env.py and exactly
    one anchor revision whose down_revision is None (a valid single-root
    chain)."""
    mig = PLUGINS_DIR / plugin / 'migrations'
    assert (mig / 'env.py').exists(), f"{plugin}: missing migrations/env.py"
    versions = sorted((mig / 'versions').glob('*.py'))
    assert versions, f"{plugin}: no version scripts"

    roots = []
    heads = set()
    down_revisions = set()
    revisions = set()
    for script in versions:
        tree = ast.parse(script.read_text())
        rev = down = None
        found_down = False
        for node in ast.walk(tree):
            if isinstance(node, ast.Assign):
                names = [t.id for t in node.targets if isinstance(t, ast.Name)]
                if 'revision' in names and isinstance(node.value, ast.Constant):
                    rev = node.value.value
                if 'down_revision' in names:
                    found_down = True
                    if isinstance(node.value, ast.Constant):
                        down = node.value.value
        assert rev, f"{plugin}: {script.name} has no revision id"
        assert found_down, f"{plugin}: {script.name} has no down_revision"
        revisions.add(rev)
        if down is None:
            roots.append(rev)
        else:
            down_revisions.add(down)
    heads = revisions - down_revisions
    assert len(roots) == 1, f"{plugin}: expected 1 root revision, got {roots}"
    assert len(heads) == 1, f"{plugin}: chain must have a single head, got {heads}"


@pytest.mark.parametrize('plugin', CUTOVER_PLUGINS)
def test_anchor_migration_is_noop(plugin):
    """The 0001 anchor's upgrade() and downgrade() are pure no-ops: no DDL
    operations, just `pass`. The core chain owns the tables at cutover.

    Scoped to CUTOVER_PLUGINS only. A plugin built after the cutover ships a
    real baseline (measuringtools), which is deliberately NOT a no-op."""
    anchor = PLUGINS_DIR / plugin / 'migrations' / 'versions' / f'0001_{plugin}_anchor.py'
    assert anchor.exists(), f"{plugin}: missing 0001 anchor migration"
    tree = ast.parse(anchor.read_text())
    funcs = {n.name: n for n in ast.walk(tree)
             if isinstance(n, ast.FunctionDef) and n.name in ('upgrade', 'downgrade')}
    assert set(funcs) == {'upgrade', 'downgrade'}, f"{plugin}: anchor missing up/downgrade"
    for name, fn in funcs.items():
        # Body may only be a docstring/comment plus a bare `pass`. No calls.
        calls = [n for n in ast.walk(fn) if isinstance(n, ast.Call)]
        assert not calls, f"{plugin}: anchor {name}() is not a no-op (found calls)"


def test_upgrade_all_on_fresh_db_is_clean_and_idempotent(tmp_path, monkeypatch):
    """`flask plugin upgrade-all` on a fresh SQLite DB (after the core schema
    is created) stamps every plugin anchor without error, and a second run is
    a no-op. Mirrors the deploy sequence: `flask db upgrade` then
    `flask plugin upgrade-all`."""
    from sqlalchemy import inspect, text
    from shopdb.config import TestingConfig
    from shopdb import create_app
    from shopdb.extensions import db
    from shopdb.plugins import plugin_manager

    db_file = tmp_path / 'fresh.db'
    url = f'sqlite:///{db_file}'

    # Point the whole app (db engine + migration manager) at one file DB so the
    # anchor stamps land where the app can read them back.
    monkeypatch.setattr(TestingConfig, 'SQLALCHEMY_DATABASE_URI', url)

    # create_app repoints the process-wide plugin_manager singleton; snapshot
    # its wiring and restore it so later tests see the session app unchanged.
    saved = (plugin_manager._app, plugin_manager._db, plugin_manager.registry,
             plugin_manager.loader, plugin_manager.migration_manager,
             plugin_manager._registered_prefixes)
    try:
        app = create_app('testing')
        with app.app_context():
            db.create_all()  # stand in for the core `flask db upgrade`

            # db.create_all() over-creates: it builds EVERY table registered on
            # the metadata, including post-cutover plugin tables the core chain
            # would never own. Drop those so each post-cutover baseline creates
            # its own tables exactly as it does after a real core upgrade (where
            # the tables are simply absent). The cutover anchors are no-ops, so
            # their create_all-built tables stay put.
            insp0 = inspect(db.engine)
            for plugin in POST_CUTOVER_PLUGINS:
                # Drop by owned name (SQLite tolerates any order with no rows);
                # avoids resolving cross-metadata FKs via sorted_tables.
                for tablename in PLUGIN_TABLE_OWNERS[plugin]:
                    if insp0.has_table(tablename):
                        db.session.execute(text(f'DROP TABLE {tablename}'))
            db.session.commit()

            first = app.extensions['plugin_manager'].upgrade_all_plugins()
            second = app.extensions['plugin_manager'].upgrade_all_plugins()

            # A plugin that owns no tables (Tech Tools: every tool is
            # client-side) carries no chain, so upgrade-all reports it
            # 'no-migrations'. Assert that explicitly rather than letting it
            # widen the 'ok' check, which would also swallow a table-owning
            # plugin whose chain silently went missing.
            schemaless = set(first) - set(PLUGIN_TABLE_OWNERS)
            for name in schemaless:
                assert first[name] == 'no-migrations', (name, first[name])
                assert second[name] == 'no-migrations', (name, second[name])

            migrated = {name: status for name, status in first.items()
                        if name in PLUGIN_TABLE_OWNERS}
            assert set(migrated) == set(PLUGIN_TABLE_OWNERS)
            assert all(status == 'ok' for status in migrated.values()), migrated
            assert all(second[name] == 'ok' for name in PLUGIN_TABLE_OWNERS), second

            insp = inspect(db.engine)
            for plugin in PLUGIN_TABLE_OWNERS:
                version_table = f'alembic_version_{plugin}'
                assert insp.has_table(version_table), f"missing {version_table}"
                row = db.session.execute(
                    text(f'SELECT version_num FROM {version_table}')
                ).fetchone()
                assert row and row[0] == EXPECTED_HEAD_REVISION[plugin]
    finally:
        (plugin_manager._app, plugin_manager._db, plugin_manager.registry,
         plugin_manager.loader, plugin_manager.migration_manager,
         plugin_manager._registered_prefixes) = saved



# Plugins whose whole chain is a single baseline. Their tables existing already
# is the ONLY divergence possible between schema and stamp, so upgrade-all must
# handle it end to end. A longer chain can also meet an ALTER whose column is
# already present, which no amount of guarding a create_table fixes - see the
# test below for why that is a different problem.
SINGLE_BASELINE_PLUGINS = tuple(
    plugin for plugin in POST_CUTOVER_PLUGINS
    if len(list((PLUGINS_DIR / plugin / 'migrations' / 'versions').glob('0*.py'))) == 1
)


@pytest.mark.parametrize('plugin', SINGLE_BASELINE_PLUGINS)
def test_baseline_skips_tables_that_already_exist(plugin, tmp_path, monkeypatch):
    """A baseline must stamp, not fail, when its table is already there.

    Reported: `flask plugin upgrade-all` failed the tools plugin with "table
    securesendsecrets already exists". Every post-cutover baseline was a plain
    op.create_table, correct only while a table's existence and its plugin's
    Alembic stamp can never disagree. They can - anything that builds tables
    from the models (db.create_all) stamps nothing, and a run that creates a
    table then fails before its stamp commits leaves the same split.

    The other upgrade-all test DROPS these tables before running, so it only
    ever covered the state where they are absent. This is the other state.
    """
    from sqlalchemy import inspect, text
    from shopdb.config import TestingConfig
    from shopdb import create_app
    from shopdb.extensions import db
    from shopdb.plugins import plugin_manager

    monkeypatch.setattr(TestingConfig, 'SQLALCHEMY_DATABASE_URI',
                        f'sqlite:///{tmp_path / "preexisting.db"}')

    saved = (plugin_manager._app, plugin_manager._db, plugin_manager.registry,
             plugin_manager.loader, plugin_manager.migration_manager,
             plugin_manager._registered_prefixes)
    try:
        app = create_app('testing')
        with app.app_context():
            db.create_all()

            # Leave THIS plugin's tables in place - that is the whole point -
            # and drop the other post-cutover ones so their own (longer) chains
            # run the way the sibling test already covers.
            insp0 = inspect(db.engine)
            for other in POST_CUTOVER_PLUGINS:
                if other == plugin:
                    continue
                for tablename in PLUGIN_TABLE_OWNERS[other]:
                    if insp0.has_table(tablename):
                        db.session.execute(text(f'DROP TABLE {tablename}'))
            db.session.commit()

            for tablename in PLUGIN_TABLE_OWNERS[plugin]:
                assert inspect(db.engine).has_table(tablename), (
                    f'{tablename} should have been built by create_all; '
                    f'this test is meaningless without it')

            first = app.extensions['plugin_manager'].upgrade_all_plugins()
            assert first[plugin] == 'ok', first[plugin]

            second = app.extensions['plugin_manager'].upgrade_all_plugins()
            assert second[plugin] == 'ok', second[plugin]

            insp = inspect(db.engine)
            row = db.session.execute(
                text(f'SELECT version_num FROM alembic_version_{plugin}')
            ).fetchone()
            assert row and row[0] == EXPECTED_HEAD_REVISION[plugin]

            # No second index over the same columns: create_all names indexes by
            # SQLAlchemy's convention and the baseline by this repo's, so a
            # name-keyed guard would have added a duplicate rather than skipped.
            for tablename in PLUGIN_TABLE_OWNERS[plugin]:
                seen = [tuple(ix.get('column_names') or ())
                        for ix in insp.get_indexes(tablename)]
                assert len(seen) == len(set(seen)), (
                    f'{tablename} has two indexes over the same columns: {seen}')
    finally:
        (plugin_manager._app, plugin_manager._db, plugin_manager.registry,
         plugin_manager.loader, plugin_manager.migration_manager,
         plugin_manager._registered_prefixes) = saved