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', 'equipment', 'knowledgebase', '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'

# 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()

            assert set(first) == set(PLUGIN_TABLE_OWNERS)
            assert all(status == 'ok' for status in first.values()), first
            assert all(status == 'ok' for status in second.values()), 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