f8c4246483
Three faults around vendor-model photos, found while looking at why an uploaded image did not appear. Saving a model was blocked after uploading a photo. The Image URL field was type="url", and an upload sets it to an application path such as /api/models/image/model-120.png. Native url validation demands an absolute URL with a scheme, so the browser refused to submit the form with "Please enter a URL" for a value the page had just written itself. The field is now type="text", which is what it always needed to be: it holds either a full web address or a path on this server. documentationurl stays type="url". The upload button did not appear when adding a model, only when editing one. That was deliberate - the photo is stored as model-<id>.<ext>, so it cannot be sent before the record has an id - but it reads as a missing feature, and the hint explaining it was easy to miss. A photo chosen while creating is now held and uploaded as soon as the model is saved, and it is dropped if the dialog is cancelled, so it cannot land on the next model created in the same session. Network devices could never show a photo. NetworkDeviceDetail.vue binds its hero image to networkdevice.imageurl, but networkdevices carried only vendorid, with no link to a catalog model, so nothing could populate it - a feature that looked present and could not work. Machines, PCs and printers have carried modelnumberid since July. This adds the same column and relationship, the to_dict branch that exposes modelname and imageurl, the field on the API, and a Model selector on the form so the link can actually be set. The migration is guarded the same way employees0002photo is: on a fresh database the tables come from the SQLAlchemy models, which already declare the column, so an unconditional add fails with "duplicate column name". The foreign key is created only on databases that can add one by ALTER; routing it through batch_alter_table made Alembic's column sort raise "Circular dependency detected" on the fresh-database test. Deploying this needs `flask db upgrade` and `flask plugin upgrade-all` on the server, not just a file copy.
253 lines
12 KiB
Python
253 lines
12 KiB
Python
"""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'
|
|
# geenforce adds the content-addressed blob store (manifestblobs) on top of its
|
|
# baseline.
|
|
EXPECTED_HEAD_REVISION['geenforce'] = 'geenforce0002blobs'
|
|
# printers adds the printersupplyalerts crossing-state table on top of its anchor.
|
|
EXPECTED_HEAD_REVISION['printers'] = 'printers0002supplyalerts'
|
|
# 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'] = 'network0002model'
|
|
# printedparts is post-cutover: its 0001 really creates its tables; 0004 adds
|
|
# the per-transaction revision column.
|
|
EXPECTED_HEAD_REVISION['printedparts'] = 'printedparts0004txnrev'
|
|
# notifications indexes businessunitid on top of its anchor.
|
|
EXPECTED_HEAD_REVISION['notifications'] = 'notifications0002buidx'
|
|
|
|
# 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
|