Merge catalog rows that are the same name typed two ways

The legacy data carried space-stripped twins: 'DellInc.' beside 'Dell Inc.',
'Windows10' beside 'Windows 10', 'OptiPlex7080' beside 'OptiPlex 7080'. They
rode into the shipped seed catalog, so every site that ran `seed catalog`
inherited them, and they stay invisible until somebody opens a vendor picker
and finds the same company listed twice.

Not a code fault. The collector strips only leading and trailing whitespace,
matches with ilike, and treats the OS list as lookup-only, so it can neither
have created these nor be creating more. They arrived with the classic-ASP
import, which holds the same three pairs today.

`flask catalog dedupe` merges only whitespace-and-punctuation twins, because
those are one name typed two ways. 'Dell' and 'Dell Inc.' normalise apart and
are left alone: that is a naming choice and only a person can make it. The
surviving row is the most referenced, tie-broken toward the spaced spelling,
which is the one somebody pressed the space bar for.

`flask catalog merge` makes that judgment call explicitly, and refuses to
invent a target that does not exist, so a typo cannot move rows into a new
vendor. `flask catalog rename` handles the case neither can reach: a stripped
name with NO twin, where there is nothing to merge into.

All three report and change nothing unless --apply is passed, and dedupe is
idempotent so it is safe in a deploy script.

References are discovered from the LIVE schema rather than listed in code.
There are 15 foreign keys into these three tables across 7 tables, and
machines.controllervendorid and printerdrivers.modelnumberid are not ones
anybody would enumerate by hand. Every reference is repointed before the row
is deleted, so nothing is orphaned, and a plugin adding a table is covered
without editing this command.
This commit is contained in:
cproudlock
2026-08-24 14:49:24 -04:00
parent 7fb6bdba07
commit d51127841d
3 changed files with 542 additions and 1 deletions

View File

@@ -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