diff --git a/tests/test_wahab_directives_20260928.py b/tests/test_wahab_directives_20260928.py index ef553a0..8bd58ee 100644 --- a/tests/test_wahab_directives_20260928.py +++ b/tests/test_wahab_directives_20260928.py @@ -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)}"