Catalog dedupe commands, and a seed catalog with no duplicate vendors or models

CHANGELOG.md

@@ -12,6 +12,21 @@ ADR-007 and ADR-002.
 
 ### Added
 
+- **`flask catalog dedupe`, `merge` and `rename`: catalog hygiene.** The legacy
+  data carried space-stripped twins - `DellInc.` beside `Dell Inc.`,
+  `Windows10` beside `Windows 10` - which rode into the shipped seed catalog
+  and out to every site that ran `seed catalog`. `dedupe` merges only
+  whitespace-and-punctuation twins, because those are one name typed two ways;
+  `Dell` and `Dell Inc.` normalise apart and are deliberately left alone, since
+  that is a naming choice only a person can make. The surviving row is the most
+  referenced, tie-broken toward the spaced spelling. `merge` makes that
+  judgment call explicitly and refuses to invent a target that does not exist.
+  `rename` fixes a stripped name with no twin, which neither of the others can
+  reach. All three report and change nothing unless `--apply` is passed, and
+  `dedupe` is idempotent. References are discovered from the live schema, not
+  listed in code: there are 15 foreign keys into these three tables, and
+  `machines.controllervendorid` is not one anybody would enumerate by hand.
+
 - **Secure Send: hand somebody a password without leaving it in a chat
   window.** A link in Tech Tools that carries one credential, dies when it is
   opened, and expires on its own if nobody opens it. The secret is encrypted in
@@ -75,6 +90,15 @@ ADR-007 and ADR-002.
 
 ### Fixed
 
+- **The shipped seed catalog contained duplicate vendors, models and operating
+  systems.** Four space-stripped twins, each with a correctly spaced original
+  already present, so a new site inherited the same company listed twice.
+  `OptiPlex5000` had no twin and was renamed to `OptiPlex 5000` rather than
+  dropped. Vendors 53 to 51, models 128 to 127, operating systems 14 to 12.
+  Not a code fault: the collector strips only leading and trailing whitespace
+  and treats the OS list as lookup-only, so it neither created these nor is
+  creating more. They came in with the classic-ASP import.
+
 - **A maintenance script could not find the database the site actually uses.**
   `python scripts/export_catalog.py` on a deployed site died with 1045 "Access
   denied for user 'root'@'localhost'", naming a user that site does not have.

shopdb/__init__.py

@@ -243,13 +243,15 @@ def register_blueprints(app: Flask):
 def register_cli_commands(app: Flask):
     """Register Flask CLI commands."""
     from .plugins.cli import plugin_cli
-    from .cli import db_cli, seed_cli, relationships_cli, csv_cli
+    from .cli import (db_cli, seed_cli, relationships_cli, csv_cli,
+                      catalog_cli)
 
     app.cli.add_command(plugin_cli)
     app.cli.add_command(db_cli)
     app.cli.add_command(seed_cli)
     app.cli.add_command(relationships_cli)
     app.cli.add_command(csv_cli)
+    app.cli.add_command(catalog_cli)
 
 
 def register_error_handlers(app: Flask):

shopdb/cli/__init__.py

@@ -1535,3 +1535,303 @@ def audit_relationships():
                    'knowing which side is authoritative.')
     else:
         click.echo(click.style('No relationship faults found.', fg='green'))
+
+
+# --- catalog hygiene --------------------------------------------------------
+
+@click.group('catalog')
+def catalog_cli():
+    """Clean up the shared vendor/model/OS catalog."""
+    pass
+
+
+# Catalog tables that carry a human-typed name, with the column that names them.
+# Each is free vocabulary somebody can add to, which is how twins appear.
+_CATALOG_TABLES = (
+    ('vendors', 'vendor'),
+    ('models', 'modelnumber'),
+    ('operatingsystems', 'osname'),
+)
+
+
+def _catalog_norm(value):
+    """Collapse a name to what it would be with no spacing or punctuation.
+
+    Only whitespace and punctuation are removed, and case is folded. That is
+    deliberately narrow: 'Dell Inc.' and 'DellInc.' are the SAME name typed two
+    ways, while 'Dell' and 'Dell Inc.' are a naming CHOICE and must never be
+    merged automatically. Use `catalog merge` for that.
+    """
+    import re
+    return re.sub(r'[\s\-_.,/]+', '', str(value or '').strip().lower())
+
+
+def _inbound_foreign_keys(inspector, table):
+    """Every (table, column) in the schema that points at `table`'s key.
+
+    Discovered from the live schema rather than listed here, because a plugin
+    can add one: `machines.controllervendorid` and `printerdrivers.
+    modelnumberid` both reference the catalog and neither is obvious.
+    """
+    refs = []
+    for candidate in inspector.get_table_names():
+        for fk in inspector.get_foreign_keys(candidate):
+            if fk['referred_table'] == table:
+                refs.append((candidate,
+                             fk['constrained_columns'][0],
+                             fk['referred_columns'][0]))
+    return refs
+
+
+def _reference_counts(connection, refs, keyvalue):
+    """How many rows across the schema point at one catalog row."""
+    import sqlalchemy as sa
+    total = 0
+    detail = []
+    for tablename, column, _ in refs:
+        count = connection.execute(
+            sa.text(f'SELECT COUNT(*) FROM {tablename} WHERE {column} = :k'),
+            {'k': keyvalue}).scalar()
+        if count:
+            detail.append(f'{tablename}.{column}={count}')
+        total += count
+    return total, detail
+
+
+def _merge_catalog_rows(connection, table, keycolumn, refs, loser, winner):
+    """Repoint every reference off `loser` onto `winner`, then delete it."""
+    import sqlalchemy as sa
+    moved = 0
+    for tablename, column, _ in refs:
+        result = connection.execute(
+            sa.text(f'UPDATE {tablename} SET {column} = :w WHERE {column} = :l'),
+            {'w': winner, 'l': loser})
+        moved += result.rowcount or 0
+    connection.execute(
+        sa.text(f'DELETE FROM {table} WHERE {keycolumn} = :l'), {'l': loser})
+    return moved
+
+
+@catalog_cli.command('dedupe')
+@click.option('--apply', 'apply_changes', is_flag=True,
+              help='Write the changes. Without this the command only reports.')
+@with_appcontext
+def catalog_dedupe(apply_changes):
+    """Merge catalog rows that are the same name typed two ways.
+
+    Legacy data arrived carrying space-stripped twins - 'DellInc.' beside
+    'Dell Inc.', 'Windows10' beside 'Windows 10' - which then rode into the
+    shipped seed catalog and out to every site that ran `seed catalog`. They
+    are invisible until somebody opens a vendor picker and finds the same
+    company twice.
+
+    ONLY whitespace-and-punctuation twins are touched. 'Dell' and 'Dell Inc.'
+    are a naming choice, not a typo, and are left alone; use `catalog merge`
+    to make that call explicitly.
+
+    The surviving row is the one with the most references, tie-broken toward
+    the spaced spelling, because that is the one a person typed deliberately.
+    Every foreign key pointing at the loser is repointed first, so nothing is
+    orphaned; references are counted across the LIVE schema, which picks up
+    plugin tables such as machines.controllervendorid.
+
+    Reports only unless --apply is passed. Idempotent.
+    """
+    import sqlalchemy as sa
+    from collections import defaultdict
+    from shopdb.extensions import db
+
+    engine = db.session.get_bind()
+    inspector = sa.inspect(engine)
+    existing = set(inspector.get_table_names())
+
+    planned = []
+    with engine.begin() as connection:
+        for table, namecolumn in _CATALOG_TABLES:
+            if table not in existing:
+                continue
+            keycolumn = inspector.get_pk_constraint(table)['constrained_columns'][0]
+            refs = [r for r in _inbound_foreign_keys(inspector, table)
+                    if r[0] in existing]
+
+            rows = connection.execute(sa.text(
+                f'SELECT {keycolumn}, {namecolumn} FROM {table}')).fetchall()
+            groups = defaultdict(list)
+            for key, name in rows:
+                if name:
+                    groups[_catalog_norm(name)].append((key, name))
+
+            for _, members in sorted(groups.items()):
+                if len(members) < 2:
+                    continue
+                scored = []
+                for key, name in members:
+                    count, detail = _reference_counts(connection, refs, key)
+                    # most-referenced wins; then the spelling with whitespace,
+                    # which is the one somebody typed on purpose; then longest.
+                    scored.append((count, ' ' in str(name), len(str(name)),
+                                   key, name, detail))
+                scored.sort(reverse=True)
+                winner = scored[0]
+                for loser in scored[1:]:
+                    planned.append({
+                        'table': table, 'keycolumn': keycolumn, 'refs': refs,
+                        'winner_key': winner[3], 'winner_name': winner[4],
+                        'loser_key': loser[3], 'loser_name': loser[4],
+                        'loser_refs': loser[0], 'loser_detail': loser[5],
+                    })
+
+        if not planned:
+            click.echo(click.style('No catalog twins found.', fg='green'))
+            return
+
+        click.echo(f'{len(planned)} duplicate row(s):')
+        click.echo('')
+        for item in planned:
+            where = (', '.join(item['loser_detail']) if item['loser_detail']
+                     else 'nothing references it')
+            click.echo(f"  {item['table']}: {item['loser_name']!r} "
+                       f"-> {item['winner_name']!r}")
+            click.echo(f"      {item['loser_refs']} reference(s): {where}")
+
+        if not apply_changes:
+            click.echo('')
+            click.echo(click.style(
+                'Dry run. Nothing was changed. Re-run with --apply to write.',
+                fg='yellow'))
+            return
+
+        moved = 0
+        for item in planned:
+            moved += _merge_catalog_rows(
+                connection, item['table'], item['keycolumn'], item['refs'],
+                item['loser_key'], item['winner_key'])
+
+    click.echo('')
+    click.echo(click.style(
+        f'Merged {len(planned)} row(s), repointed {moved} reference(s).',
+        fg='green'))
+
+
+@catalog_cli.command('merge')
+@click.argument('table', type=click.Choice([t for t, _ in _CATALOG_TABLES]))
+@click.argument('source')
+@click.argument('target')
+@click.option('--apply', 'apply_changes', is_flag=True,
+              help='Write the changes. Without this the command only reports.')
+@with_appcontext
+def catalog_merge(table, source, target, apply_changes):
+    """Merge one catalog row into another by name: SOURCE is absorbed by TARGET.
+
+    For the judgment calls `catalog dedupe` deliberately refuses to make -
+    'Dell' and 'Dell Inc.' are different names, and only a person knows the
+    site means one company.
+
+        flask catalog merge vendors "Dell" "Dell Inc." --apply
+
+    Every reference to SOURCE is repointed to TARGET, then SOURCE is deleted.
+    Reports only unless --apply is passed.
+    """
+    import sqlalchemy as sa
+    from shopdb.extensions import db
+
+    namecolumn = dict(_CATALOG_TABLES)[table]
+    engine = db.session.get_bind()
+    inspector = sa.inspect(engine)
+    existing = set(inspector.get_table_names())
+    keycolumn = inspector.get_pk_constraint(table)['constrained_columns'][0]
+    refs = [r for r in _inbound_foreign_keys(inspector, table) if r[0] in existing]
+
+    with engine.begin() as connection:
+        def lookup(value):
+            return connection.execute(
+                sa.text(f'SELECT {keycolumn} FROM {table} '
+                        f'WHERE {namecolumn} = :v'), {'v': value}).fetchone()
+
+        sourcerow, targetrow = lookup(source), lookup(target)
+        if not sourcerow:
+            click.echo(click.style(f'No {table} row named {source!r}.', fg='yellow'))
+            return
+        if not targetrow:
+            click.echo(click.style(f'No {table} row named {target!r}. '
+                                   f'Refusing to invent it.', fg='red'))
+            raise SystemExit(1)
+        if sourcerow[0] == targetrow[0]:
+            click.echo(click.style('Source and target are the same row.', fg='yellow'))
+            return
+
+        count, detail = _reference_counts(connection, refs, sourcerow[0])
+        where = ', '.join(detail) if detail else 'nothing references it'
+        click.echo(f'{table}: {source!r} -> {target!r}')
+        click.echo(f'    {count} reference(s): {where}')
+
+        if not apply_changes:
+            click.echo(click.style(
+                'Dry run. Nothing was changed. Re-run with --apply to write.',
+                fg='yellow'))
+            return
+
+        moved = _merge_catalog_rows(connection, table, keycolumn, refs,
+                                    sourcerow[0], targetrow[0])
+
+    click.echo(click.style(
+        f'Merged {source!r} into {target!r}, repointed {moved} reference(s).',
+        fg='green'))
+
+
+@catalog_cli.command('rename')
+@click.argument('table', type=click.Choice([t for t, _ in _CATALOG_TABLES]))
+@click.argument('source')
+@click.argument('target')
+@click.option('--apply', 'apply_changes', is_flag=True,
+              help='Write the change. Without this the command only reports.')
+@with_appcontext
+def catalog_rename(table, source, target, apply_changes):
+    """Rename one catalog row in place, keeping every reference to it.
+
+    A space-stripped name with NO twin cannot be deduped - there is nothing to
+    merge it into - so 'OptiPlex5000' would survive a `catalog dedupe` that
+    correctly removed 'OptiPlex7080'. This fixes that case.
+
+        flask catalog rename models "OptiPlex5000" "OptiPlex 5000" --apply
+
+    Refuses when the target name already exists, because that is a merge and
+    has to move references rather than rename a row: use `catalog merge`.
+    """
+    import sqlalchemy as sa
+    from shopdb.extensions import db
+
+    namecolumn = dict(_CATALOG_TABLES)[table]
+    engine = db.session.get_bind()
+    inspector = sa.inspect(engine)
+    keycolumn = inspector.get_pk_constraint(table)['constrained_columns'][0]
+
+    with engine.begin() as connection:
+        def lookup(value):
+            return connection.execute(
+                sa.text(f'SELECT {keycolumn} FROM {table} '
+                        f'WHERE {namecolumn} = :v'), {'v': value}).fetchone()
+
+        sourcerow = lookup(source)
+        if not sourcerow:
+            click.echo(click.style(f'No {table} row named {source!r}.', fg='yellow'))
+            return
+        if lookup(target):
+            click.echo(click.style(
+                f'{table} already has a row named {target!r}. That is a merge, '
+                f'not a rename - use `catalog merge`.', fg='red'))
+            raise SystemExit(1)
+
+        click.echo(f'{table}: rename {source!r} -> {target!r} '
+                   f'(id {sourcerow[0]}, references unaffected)')
+        if not apply_changes:
+            click.echo(click.style(
+                'Dry run. Nothing was changed. Re-run with --apply to write.',
+                fg='yellow'))
+            return
+
+        connection.execute(
+            sa.text(f'UPDATE {table} SET {namecolumn} = :t WHERE {keycolumn} = :k'),
+            {'t': target, 'k': sourcerow[0]})
+
+    click.echo(click.style(f'Renamed {source!r} to {target!r}.', fg='green'))

shopdb/data/catalog.json

@@ -383,9 +383,6 @@
     {
       "osname": "Microsoft Windows 11 Enterprise"
     },
-    {
-      "osname": "MicrosoftWindows10Enterprise"
-    },
     {
       "osname": "TBD"
     },
@@ -404,9 +401,6 @@
     {
       "osname": "Windows Server 2022",
       "osversion": "2022"
-    },
-    {
-      "osname": "Windows10"
     }
   ],
   "vendors": [
@@ -432,15 +426,9 @@
     {
       "vendor": "Cisco"
     },
-    {
-      "vendor": "Dell"
-    },
     {
       "vendor": "Dell Inc."
     },
-    {
-      "vendor": "DellInc."
-    },
     {
       "vendor": "DMG Mori"
     },
@@ -620,8 +608,8 @@
       "modeltype": "Switch"
     },
     {
-      "modelnumber": "OptiPlex5000",
+      "modelnumber": "OptiPlex 5000",
-      "vendor": "Dell"
+      "vendor": "Dell Inc."
     },
     {
       "modelnumber": "Dell Pro 13 Plus PB13250",
@@ -775,10 +763,6 @@
       "vendor": "Dell Inc.",
       "modeltype": "LocationOnly"
     },
-    {
-      "modelnumber": "OptiPlex7080",
-      "vendor": "DellInc."
-    },
     {
       "modelnumber": "NT-4300",
       "vendor": "DMG Mori",

tests/test_core/test_catalog_dedupe.py

@@ -0,0 +1,239 @@
+"""Catalog hygiene: merging vendor/model/OS rows that are the same name twice.
+
+Legacy data arrived carrying space-stripped twins - 'DellInc.' beside
+'Dell Inc.', 'Windows10' beside 'Windows 10'. They rode into the shipped seed
+catalog and out to every site that ran `seed catalog`, and stay invisible until
+somebody opens a vendor picker and finds the same company listed twice.
+
+The distinction these tests exist to protect: a twin is the same name TYPED two
+ways and may be merged mechanically, while 'Dell' and 'Dell Inc.' are two
+different names and merging them is a judgment only a person can make.
+"""
+
+import pytest
+
+from shopdb.extensions import db
+from shopdb.core.models import Vendor, Model
+from shopdb.cli import _catalog_norm
+
+
+def _run(app, args):
+    return app.test_cli_runner().invoke(args=args)
+
+
+# --- the normaliser draws the line ------------------------------------------
+
+def test_whitespace_and_punctuation_twins_normalise_together():
+    assert _catalog_norm('Dell Inc.') == _catalog_norm('DellInc.')
+    assert _catalog_norm('Windows 10') == _catalog_norm('Windows10')
+    assert _catalog_norm('OptiPlex 7080') == _catalog_norm('OptiPlex7080')
+
+
+def test_a_different_name_does_not_normalise_together():
+    """'Dell' is not a typo of 'Dell Inc.', it is a shorter name.
+
+    If these collapsed, `dedupe` would silently pick one and delete a name the
+    site may have chosen deliberately.
+    """
+    assert _catalog_norm('Dell') != _catalog_norm('Dell Inc.')
+    assert _catalog_norm('Precision 5560') != _catalog_norm('Precision 5570')
+
+
+# --- dedupe ------------------------------------------------------------------
+
+@pytest.fixture
+def dell_twins(db):
+    """The real shape found in production: a twin, plus a genuine third name."""
+    canonical = Vendor(vendor='Dell Inc.')
+    twin = Vendor(vendor='DellInc.')
+    shortname = Vendor(vendor='Dell')
+    db.session.add_all([canonical, twin, shortname])
+    db.session.flush()
+    # the twin owns a model, so the merge has a reference to repoint
+    db.session.add(Model(modelnumber='OptiPlex7080', vendorid=twin.vendorid))
+    db.session.add(Model(modelnumber='OptiPlex 7080', vendorid=canonical.vendorid))
+    db.session.commit()
+    return {'canonical': canonical.vendorid, 'twin': twin.vendorid,
+            'shortname': shortname.vendorid}
+
+
+def test_dedupe_reports_without_changing_anything(app, dell_twins):
+    """The default is a dry run: a cleanup that writes by surprise is a trap."""
+    result = _run(app, ['catalog', 'dedupe'])
+
+    assert result.exit_code == 0
+    assert 'DellInc.' in result.output
+    assert 'Dry run' in result.output
+    # every row still stands
+    assert db.session.get(Vendor, dell_twins['twin']) is not None
+    assert Vendor.query.filter_by(vendor='DellInc.').count() == 1
+
+
+def test_dedupe_merges_the_twin_and_repoints_its_model(app, dell_twins):
+    """The twin disappears and its model moves, rather than being orphaned."""
+    result = _run(app, ['catalog', 'dedupe', '--apply'])
+
+    assert result.exit_code == 0
+    assert Vendor.query.filter_by(vendor='DellInc.').count() == 0
+    assert Vendor.query.filter_by(vendor='Dell Inc.').count() == 1
+
+    # nothing points at a vendor that no longer exists
+    vendorids = {v.vendorid for v in Vendor.query.all()}
+    for model in Model.query.all():
+        assert model.vendorid is None or model.vendorid in vendorids
+
+
+def test_dedupe_leaves_a_genuinely_different_name_alone(app, dell_twins):
+    """'Dell' survives. It is a naming choice, not a typo."""
+    _run(app, ['catalog', 'dedupe', '--apply'])
+
+    assert Vendor.query.filter_by(vendor='Dell').count() == 1
+
+
+def test_dedupe_keeps_the_spelling_a_person_typed(app, db):
+    """With no references either way, the spaced spelling wins.
+
+    Whitespace is evidence of intent: somebody pressed the space bar.
+    """
+    db.session.add_all([Vendor(vendor='AcmeCorp'), Vendor(vendor='Acme Corp')])
+    db.session.commit()
+
+    _run(app, ['catalog', 'dedupe', '--apply'])
+
+    assert Vendor.query.filter_by(vendor='Acme Corp').count() == 1
+    assert Vendor.query.filter_by(vendor='AcmeCorp').count() == 0
+
+
+def test_dedupe_keeps_the_most_referenced_row(app, db):
+    """References outrank spelling: the row in use is the one that survives."""
+    bare = Vendor(vendor='ZZTop')
+    spaced = Vendor(vendor='ZZ Top')
+    db.session.add_all([bare, spaced])
+    db.session.flush()
+    for i in range(3):
+        db.session.add(Model(modelnumber=f'ZZ-{i}', vendorid=bare.vendorid))
+    db.session.commit()
+
+    _run(app, ['catalog', 'dedupe', '--apply'])
+
+    assert Vendor.query.filter_by(vendor='ZZTop').count() == 1
+    assert Vendor.query.filter_by(vendor='ZZ Top').count() == 0
+
+
+def test_dedupe_is_idempotent(app, dell_twins):
+    """Running it twice is a no-op, so it is safe in a deploy script."""
+    _run(app, ['catalog', 'dedupe', '--apply'])
+    second = _run(app, ['catalog', 'dedupe', '--apply'])
+
+    assert second.exit_code == 0
+    assert 'No catalog twins found' in second.output
+
+
+def test_dedupe_on_a_clean_catalog_says_so(app, db):
+    db.session.add(Vendor(vendor='Solo Vendor'))
+    db.session.commit()
+
+    result = _run(app, ['catalog', 'dedupe'])
+
+    assert result.exit_code == 0
+    assert 'No catalog twins found' in result.output
+
+
+# --- explicit merge ----------------------------------------------------------
+
+def test_merge_absorbs_a_deliberately_different_name(app, dell_twins):
+    """The judgment call dedupe refuses: 'Dell' means 'Dell Inc.' at this site."""
+    result = _run(app, ['catalog', 'merge', 'vendors', 'Dell', 'Dell Inc.',
+                        '--apply'])
+
+    assert result.exit_code == 0
+    assert Vendor.query.filter_by(vendor='Dell').count() == 0
+    assert Vendor.query.filter_by(vendor='Dell Inc.').count() == 1
+
+
+def test_merge_dry_runs_by_default(app, dell_twins):
+    result = _run(app, ['catalog', 'merge', 'vendors', 'Dell', 'Dell Inc.'])
+
+    assert 'Dry run' in result.output
+    assert Vendor.query.filter_by(vendor='Dell').count() == 1
+
+
+def test_merge_refuses_to_invent_the_target(app, dell_twins):
+    """A typo in the target must not create a new vendor and move rows into it."""
+    result = _run(app, ['catalog', 'merge', 'vendors', 'Dell', 'Dell Incorporated',
+                        '--apply'])
+
+    assert result.exit_code == 1
+    assert 'Refusing to invent' in result.output
+    assert Vendor.query.filter_by(vendor='Dell').count() == 1
+
+
+def test_merge_reports_a_missing_source_without_failing(app, dell_twins):
+    """Already merged is not an error: the deploy script can just run again."""
+    result = _run(app, ['catalog', 'merge', 'vendors', 'Nonexistent',
+                        'Dell Inc.', '--apply'])
+
+    assert result.exit_code == 0
+    assert 'No vendors row named' in result.output
+
+
+# --- rename ------------------------------------------------------------------
+
+def test_rename_fixes_a_stripped_name_with_no_twin(app, db):
+    """'OptiPlex5000' has nothing to merge into, so dedupe cannot reach it."""
+    vendor = Vendor(vendor='Dell Inc.')
+    db.session.add(vendor)
+    db.session.flush()
+    db.session.add(Model(modelnumber='OptiPlex5000', vendorid=vendor.vendorid))
+    db.session.commit()
+
+    _run(app, ['catalog', 'dedupe', '--apply'])
+    assert Model.query.filter_by(modelnumber='OptiPlex5000').count() == 1, \
+        'dedupe should not touch a name with no twin'
+
+    result = _run(app, ['catalog', 'rename', 'models', 'OptiPlex5000',
+                        'OptiPlex 5000', '--apply'])
+
+    assert result.exit_code == 0
+    assert Model.query.filter_by(modelnumber='OptiPlex 5000').count() == 1
+    assert Model.query.filter_by(modelnumber='OptiPlex5000').count() == 0
+
+
+def test_rename_keeps_the_row_and_its_references(app, db):
+    """A rename is in place: the id does not change, so nothing is repointed."""
+    vendor = Vendor(vendor='Acme')
+    db.session.add(vendor)
+    db.session.flush()
+    model = Model(modelnumber='Widget100', vendorid=vendor.vendorid)
+    db.session.add(model)
+    db.session.commit()
+    original = model.modelnumberid
+
+    _run(app, ['catalog', 'rename', 'models', 'Widget100', 'Widget 100', '--apply'])
+
+    renamed = Model.query.filter_by(modelnumber='Widget 100').one()
+    assert renamed.modelnumberid == original
+    assert renamed.vendorid == vendor.vendorid
+
+
+def test_rename_refuses_when_the_target_already_exists(app, db):
+    """Renaming onto an existing name would need a merge, not an UPDATE."""
+    db.session.add_all([Vendor(vendor='Acme'), Vendor(vendor='Acme Corp')])
+    db.session.commit()
+
+    result = _run(app, ['catalog', 'rename', 'vendors', 'Acme', 'Acme Corp',
+                        '--apply'])
+
+    assert result.exit_code == 1
+    assert 'use `catalog merge`' in result.output
+    assert Vendor.query.filter_by(vendor='Acme').count() == 1
+
+
+def test_rename_dry_runs_by_default(app, db):
+    db.session.add(Vendor(vendor='Acme'))
+    db.session.commit()
+
+    result = _run(app, ['catalog', 'rename', 'vendors', 'Acme', 'Acme Corp'])
+
+    assert 'Dry run' in result.output
+    assert Vendor.query.filter_by(vendor='Acme').count() == 1