Merge pull request 'fix(penthouse): consolidate duplicates when both legacy and clean rows exist' (#19) from fm/denya-penthouse-fix into main

app/services/seed.py

@@ -232,10 +232,9 @@ async def seed_units(db: AsyncSession, json_path: str | Path | None = None) -> l
     # Self-heal malformed penthouse codes from earlier mappings
     # (``PH1E-`` with a dangling dash). Client directive 2026-09-28: the
     # penthouse units (Pent House East/West) must be selectable when raising
-    # a ticket — they were present but mangled. Rename in place so any
+    # a ticket — they were present but mangled. Normalize to exactly one
-    # ``tickets.unit_id`` FKs stay intact; only rename when the clean code
+    # row per unit (no duplicates) while preserving ticket references.
-    # is free, and deactivate (never delete) a row that would collide.
+    consolidated = 0
-    renamed = 0
     for legacy_code, clean_code in _LEGACY_PENTHOUSE_RENAMES.items():
         legacy_row = (
             await db.execute(select(Unit).where(Unit.apartment_code == legacy_code))
@@ -245,17 +244,36 @@ async def seed_units(db: AsyncSession, json_path: str | Path | None = None) -> l
         clash = (
             await db.execute(select(Unit).where(Unit.apartment_code == clean_code))
         ).scalar_one_or_none()
-        if clash is not None:
+        if clash is None:
-            # A clean-code row already exists; leave the legacy row as an
+            # Only legacy row exists: rename in place so tickets keep their
-            # inert historical alias (its tickets keep pointing at it).
+            # FK target.
+            legacy_row.apartment_code = clean_code
+            if legacy_row.floor is None:
+                legacy_row.floor = _PENTHOUSE_FLOORS.get(clean_code)
+            consolidated += 1
             continue
-        legacy_row.apartment_code = clean_code
+        # --- Both forms exist (live condition) ---
+        # Consolidate to exactly one row. The clean-code row is the canonical
+        # survivor (correct code and deterministic floor). Re-point any
+        # tickets from the legacy row to the clean row, then remove the
+        # legacy row. Idempotent: re-pointing already-correct refs is a no-op.
+        from app.models.ticket import Ticket
         if legacy_row.floor is None:
-            legacy_row.floor = _PENTHOUSE_FLOORS.get(clean_code)
+            legacy_row.floor = _PENTHOUSE_FLOORS.get(legacy_code)
-        renamed += 1
+        tickets_on_legacy = (
-    if renamed:
+            await db.execute(
+                select(Ticket).where(Ticket.unit_id == legacy_row.id)
+            )
+        ).scalars().all()
+        for ticket in tickets_on_legacy:
+            ticket.unit_id = clash.id
+        await db.delete(legacy_row)
+        if clash.floor is None:
+            clash.floor = _PENTHOUSE_FLOORS.get(clean_code)
+        consolidated += 1
+    if consolidated:
         await db.flush()
-        logger.info("Repaired %d legacy penthouse unit codes", renamed)
+        logger.info("Consolidated %d legacy penthouse rows (renamed or deduped)", consolidated)
 
     if created:
         await db.flush()

tests/test_wahab_directives_20260928.py

@@ -195,3 +195,83 @@ async def test_penthouse_units_have_floors(client):
     ]
     assert len(penthouses) == 4
     assert all(u["floor"] is not None for u in penthouses)
+
+
+async def test_penthouse_consolidation_when_both_forms_exist(client, db):
+    """Regression: when BOTH legacy (PH1E-) and clean (PH1E) rows exist,
+    seed_units consolidates to exactly one row per unit and re-points
+    tickets. This reproduces the live condition on CT 115 where the
+    mapping file was cleaned but legacy rows were never removed.
+
+    Steps:
+    1. `client` fixture seeds clean rows (PH1E, PH1W, PH2E, PH2W with floors).
+    2. We insert the legacy malformed rows (PH1E- etc.) with floor=null.
+    3. We create a ticket referencing one of the legacy rows.
+    4. We re-run `seed_units` to trigger the self-heal.
+    5. We assert: exactly 4 penthouse units, no dangling dashes, all have
+       floors, and the ticket still references a valid unit (the clean row).
+    """
+    from sqlalchemy import select
+    from app.core.database import engine
+    from app.models.unit import Unit
+    from app.services.seed import seed_units
+
+    # Step 1: `client` fixture already seeded clean rows (PH1E etc.)
+    # Step 2: Insert legacy malformed rows to simulate the live condition.
+    legacy_rows = [
+        Unit(apartment_code="PH1E-", property="East", building="Pavilion East", floor=None),
+        Unit(apartment_code="PH1W-", property="West", building="Pavilion West", floor=None),
+        Unit(apartment_code="PH2E-", property="East", building="Pavilion East", floor=None),
+        Unit(apartment_code="PH2W-", property="West", building="Pavilion West", floor=None),
+    ]
+    for unit in legacy_rows:
+        db.add(unit)
+    await db.flush()
+
+    # Step 3: Create a ticket referencing one of the legacy rows.
+    legacy_unit = legacy_rows[0]  # PH1E-
+    ticket = Ticket(
+        ticket_number="PAV-CONSOLIDATION-TEST",
+        status="Logged",
+        priority="medium",
+        description="Regression: ticket referencing legacy penthouse row",
+    )
+    ticket.unit_id = legacy_unit.id
+    db.add(ticket)
+    await db.flush()
+    ticket_id = ticket.id
+
+    # Step 4: Re-run seed_units to trigger the self-heal.
+    await seed_units(db, json_path=None)
+
+    # Step 5: Assert the consolidation result.
+    resp = await client.get("/api/tickets/units/grouped")
+    assert resp.status_code == 200
+    grouped = resp.json()
+
+    # Exactly 4 penthouse units (no duplicates).
+    penthouses = [
+        u
+        for wing in grouped.values()
+        for units in wing.values()
+        for u in units
+        if u["apartment_code"].startswith("PH")
+    ]
+    assert len(penthouses) == 4, f"Expected 4 penthouse units, got {len(penthouses)}"
+
+    # No dangling dashes.
+    codes = _grouped_codes(grouped)
+    assert not any(c.endswith("-") for c in codes), "no dangling-dash codes after consolidation"
+
+    # All have floors.
+    assert all(u["floor"] is not None for u in penthouses)
+
+    # The ticket still references a valid unit (the clean row).
+    reloaded = await db.execute(select(Ticket).where(Ticket.id == ticket_id))
+    reloaded_ticket = reloaded.scalar_one()
+    assert reloaded_ticket.unit_id is not None
+    # The unit_id should now be the clean row's ID (not the deleted legacy row).
+    clean_row = (
+        await db.execute(select(Unit).where(Unit.apartment_code == "PH1E"))
+    ).scalar_one_or_none()
+    assert reloaded_ticket.unit_id == clean_row.id, "ticket should reference the clean PH1E row"