test(penthouse): add idempotency and multi-ticket regression tests

Closes the two coverage gaps from the parent's 057.msg: - test_penthouse_consolidation_is_idempotent: second seed_units pass is a no-op - test_penthouse_consolidation_repoints_multiple_tickets: all tickets on legacy row re-pointed - Meta-tests prove the idempotency assertion catches delete/duplicate bugs
2026-10-02 13:46:25 +00:00
parent 0449390cbb
commit 6ce42f327e
1 changed files with 162 additions and 0 deletions
@@ -275,3 +275,165 @@ async def test_penthouse_consolidation_when_both_forms_exist(client, db):
        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"
+
+
+async def test_penthouse_consolidation_is_idempotent(client, db):
+    """Idempotency: a second seed_units pass after consolidation changes
+    nothing — still exactly four penthouse rows, same ids, same floors,
+    same ticket references. Catches a second pass that would delete or
+    duplicate a row."""
+    from sqlalchemy import select
+    from app.models.unit import Unit
+    from app.services.seed import seed_units
+
+    # Set up the live condition: clean rows exist (fixture), add legacy rows.
+    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()
+
+    # A ticket on the legacy PH1E- row.
+    ticket = Ticket(ticket_number="PAV-IDEMPOTENT", status="Logged", priority="medium", description="idempotency")
+    ticket.unit_id = legacy_rows[0].id
+    db.add(ticket)
+    await db.flush()
+    ticket_id = ticket.id
+
+    async def snapshot():
+        rows = (await db.execute(select(Unit).where(Unit.apartment_code.like("PH%")))).scalars().all()
+        state = {
+            "count": len(rows),
+            "ids": sorted(r.id for r in rows),
+            "floors": {r.apartment_code: r.floor for r in rows},
+        }
+        tk = (await db.execute(select(Ticket).where(Ticket.id == ticket_id))).scalar_one()
+        state["ticket_unit"] = tk.unit_id
+        return state
+
+    # First pass: consolidation runs.
+    await seed_units(db, json_path=None)
+    before = await snapshot()
+
+    # Second pass: idempotency check.
+    await seed_units(db, json_path=None)
+    after = await snapshot()
+
+    # Assert nothing changed.
+    assert after["count"] == 4, f"Expected 4 penthouse rows after second pass, got {after['count']}"
+    assert after["ids"] == before["ids"], "Penthouse row IDs changed between passes"
+    assert after["floors"] == before["floors"], "Floors changed between passes"
+    assert after["ticket_unit"] == before["ticket_unit"], "Ticket unit_id changed between passes"
+
+
+async def test_penthouse_consolidation_repoints_multiple_tickets(client, db):
+    """When the legacy row has multiple tickets, ALL are re-pointed to
+    the clean row during consolidation, not just the one referenced in the
+    single-ticket test."""
+    from sqlalchemy import select
+    from app.models.unit import Unit
+    from app.services.seed import seed_units
+
+    # Set up: clean rows exist (fixture), add legacy row.
+    legacy_row = Unit(apartment_code="PH1E-", property="East", building="Pavilion East", floor=None)
+    db.add(legacy_row)
+    await db.flush()
+
+    # TWO tickets on the legacy row.
+    ticket1 = Ticket(ticket_number="PAV-MT-1", status="Logged", priority="medium", description="multi-ticket 1")
+    ticket1.unit_id = legacy_row.id
+    db.add(ticket1)
+    await db.flush()
+
+    ticket2 = Ticket(ticket_number="PAV-MT-2", status="Logged", priority="medium", description="multi-ticket 2")
+    ticket2.unit_id = legacy_row.id
+    db.add(ticket2)
+    await db.flush()
+
+    ticket1_id = ticket1.id
+    ticket2_id = ticket2.id
+
+    # Consolidation.
+    await seed_units(db, json_path=None)
+
+    # Re-pointed to the clean PH1E row.
+    clean_row = (await db.execute(select(Unit).where(Unit.apartment_code == "PH1E"))).scalar_one()
+    for t_id, label in [(ticket1_id, "PAV-MT-1"), (ticket2_id, "PAV-MT-2")]:
+        t = (await db.execute(select(Ticket).where(Ticket.id == t_id))).scalar_one()
+        assert t.unit_id == clean_row.id, f"{label} should reference clean PH1E, not legacy PH1E-"
+
+
+# --- Meta-tests: prove the idempotency assertion is meaningful --------------
+# These run a real first seed_units pass, capture a snapshot, then simulate a
+# buggy second pass (delete / duplicate a clean row) and show the SAME
+# assertion logic from test_penthouse_consolidation_is_idempotent catches both.
+
+
+async def test_idempotency_assertion_catches_delete(client, db):
+    """If a second pass deleted a clean row, the count==4 assertion fails."""
+    from sqlalchemy import select
+    from app.models.unit import Unit
+    from app.services.seed import seed_units
+
+    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 u in legacy_rows:
+        db.add(u)
+    await db.flush()
+    ticket = Ticket(ticket_number="PAV-META-DEL", status="Logged", priority="medium", description="meta delete")
+    ticket.unit_id = legacy_rows[0].id
+    db.add(ticket)
+    await db.flush()
+
+    await seed_units(db, json_path=None)
+
+    # Simulate a buggy second pass that DELETES a clean row.
+    row = (await db.execute(select(Unit).where(Unit.apartment_code == "PH1E"))).scalar_one()
+    await db.delete(row)
+    await db.flush()
+
+    rows = (await db.execute(select(Unit).where(Unit.apartment_code.like("PH%")))).scalars().all()
+    with pytest.raises(AssertionError):
+        assert len(rows) == 4, f"Expected 4 penthouse rows, got {len(rows)}"
+
+
+async def test_idempotency_assertion_catches_duplicate(client, db):
+    """If a second pass added a new penthouse row, the count==4 assertion fails.
+    Same-code duplicates are blocked by the UNIQUE constraint on apartment_code.
+    This test simulates a bug that adds a new distinct penthouse code (PH3E)."""
+    from sqlalchemy import select
+    from app.models.unit import Unit
+    from app.services.seed import seed_units
+
+    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 u in legacy_rows:
+        db.add(u)
+    await db.flush()
+    ticket = Ticket(ticket_number="PAV-META-DUP", status="Logged", priority="medium", description="meta duplicate")
+    ticket.unit_id = legacy_rows[0].id
+    db.add(ticket)
+    await db.flush()
+
+    await seed_units(db, json_path=None)
+
+    # Simulate a buggy second pass that ADDS a new distinct penthouse row (PH3E).
+    # Same-code duplicates are blocked by the UNIQUE constraint on apartment_code.
+    db.add(Unit(apartment_code="PH3E", property="East", building="Pavilion East", floor=3))
+    await db.flush()
+
+    rows = (await db.execute(select(Unit).where(Unit.apartment_code.like("PH%")))).scalars().all()
+    with pytest.raises(AssertionError):
+        assert len(rows) == 4, f"Expected 4 penthouse rows, got {len(rows)}"