Phase 7A fix: BINARY comparison on MySQL for ADR-001 type seeding

MySQL's default collation (utf8mb4_general_ci) is case-insensitive, so
`WHERE relationshiptype = 'controls'` matched a legacy `Controls` row.
The check skipped the insert of the lowercase ADR-001 type, then the
follow-up UPDATE accidentally wired propagatesthroughid onto the legacy
capitalized row instead of the new canonical one.

Surfaced in live dev DB after running `flask db upgrade`:
- partof, connectedto inserted correctly
- controls NOT inserted (collision with legacy `Controls`)
- legacy `Controls` row got propagation FK wired by mistake

Fix uses BINARY comparison on MySQL in both paths:
- migrations/versions/7a01_adr001_position_contract.py: dialect-aware
  _eq() helper wraps each WHERE clause in BINARY when on MySQL. SQLite
  and PostgreSQL stay case-sensitive by default; the plain comparison
  is safe there.
- shopdb/cli/__init__.py: same dialect-aware _lookup_binary() using
  func.binary() in the SQLAlchemy query.

Dev DB healed manually by renaming `Controls` -> `controls` and wiring
propagatesthroughid to partof. Other deployments that ran the buggy
migration need the same one-line UPDATE:
    UPDATE relationshiptypes
    SET relationshiptype = 'controls', propagatesthroughid = <partof_id>
    WHERE relationshiptype = 'Controls';
(only if the deployment had a legacy capitalized row; fresh DBs are fine).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
cproudlock
2026-05-30 14:47:16 -04:00
parent 689f1a21e2
commit f419db94e5
2 changed files with 36 additions and 8 deletions

View File

@@ -70,14 +70,28 @@ def upgrade():
)
conn = op.get_bind()
dialect_name = conn.dialect.name
# MySQL default collation is case-insensitive, which makes a plain
# `WHERE relationshiptype = 'controls'` match a legacy `Controls` row
# and silently skip the insert (then accidentally wire propagation onto
# the wrong row). Force binary comparison on MySQL so the three ADR
# types are always distinct from legacy capitalized names. SQLite +
# PostgreSQL are case-sensitive by default; the plain comparison is
# safe there.
def _eq(col_expr, value):
if dialect_name == 'mysql':
return f"BINARY {col_expr} = '{value}'"
return f"{col_expr} = '{value}'"
for rt in (
('partof', 'Composition / sub-assembly (ADR-001)'),
('controls', 'Operational authority over another asset (ADR-001)'),
('connectedto', 'Network or data link without authority (ADR-001)'),
):
existing = conn.execute(sa.text(
"SELECT relationshiptypeid FROM relationshiptypes WHERE relationshiptype = :n"
), {'n': rt[0]}).first()
f"SELECT relationshiptypeid FROM relationshiptypes WHERE {_eq('relationshiptype', rt[0])}"
)).first()
if not existing:
conn.execute(relationshiptypes.insert().values(
relationshiptype=rt[0],
@@ -87,11 +101,11 @@ def upgrade():
))
partof_row = conn.execute(sa.text(
"SELECT relationshiptypeid FROM relationshiptypes WHERE relationshiptype = 'partof'"
f"SELECT relationshiptypeid FROM relationshiptypes WHERE {_eq('relationshiptype', 'partof')}"
)).first()
if partof_row:
conn.execute(sa.text(
"UPDATE relationshiptypes SET propagatesthroughid = :p WHERE relationshiptype = 'controls'"
f"UPDATE relationshiptypes SET propagatesthroughid = :p WHERE {_eq('relationshiptype', 'controls')}"
), {'p': partof_row[0]})

View File

@@ -117,21 +117,35 @@ def seed_reference_data():
# ADR-001 canonical relationship types. Created first without propagation
# FKs, then patched with propagatesthroughid since `controls` points at
# `partof` (same table). All three are idempotent.
#
# MySQL collation is case-insensitive by default, which would let a
# legacy capitalized row (e.g. "Controls") match the lowercase
# "controls" check and skip the insert. Force binary comparison via
# collate so the three ADR-001 types stay distinct from any legacy
# rows with the same spelling but different case.
from sqlalchemy import func, literal
def _lookup_binary(name):
dialect = db.engine.dialect.name
if dialect == 'mysql':
return RelationshipType.query.filter(
func.binary(RelationshipType.relationshiptype) == literal(name)
).first()
return RelationshipType.query.filter_by(relationshiptype=name).first()
adr_types = [
{'relationshiptype': 'partof', 'description': 'Composition / sub-assembly (ADR-001)'},
{'relationshiptype': 'controls', 'description': 'Operational authority over another asset (ADR-001)'},
{'relationshiptype': 'connectedto', 'description': 'Network or data link without authority (ADR-001)'},
]
for at in adr_types:
existing = RelationshipType.query.filter_by(relationshiptype=at['relationshiptype']).first()
if not existing:
if not _lookup_binary(at['relationshiptype']):
db.session.add(RelationshipType(**at))
db.session.flush()
# Wire `controls` -> `partof` propagation rail. partof + connectedto stay
# null (no propagation).
partof = RelationshipType.query.filter_by(relationshiptype='partof').first()
controls = RelationshipType.query.filter_by(relationshiptype='controls').first()
partof = _lookup_binary('partof')
controls = _lookup_binary('controls')
if partof and controls and controls.propagatesthroughid != partof.relationshiptypeid:
controls.propagatesthroughid = partof.relationshiptypeid