from datetime import UTC, datetime from sqlalchemy import func, select from app.core.db import SessionLocal from app.models.audit import AuditEvent from app.models.booking import Booking from app.models.customer import Customer from app.models.data_quality import DataQualityIssue from app.models.outbox import OutboxEvent from app.models.user import User from app.models.vehicle import Vehicle from app.seed_loader import SEED_AUTHORED_ANCHOR, reset_and_seed def test_seed_counts_match_deterministic_dataset(): # Other test modules mutate shared demo state (returns, resets), so this test # re-seeds immediately before asserting counts rather than trusting whatever # order pytest happened to run modules in. db = SessionLocal() try: reset_and_seed(db) assert db.scalar(select(func.count()).select_from(Vehicle)) == 50 assert db.scalar(select(func.count()).select_from(Customer)) == 180 assert db.scalar(select(func.count()).select_from(Booking)) == 246 # 15 from the CSV plus a deterministic set discovered by the post-seed scan. assert db.scalar(select(func.count()).select_from(DataQualityIssue)) == 26 assert db.scalar(select(func.count()).select_from(OutboxEvent)) == 20 assert db.scalar(select(func.count()).select_from(User)) == 2 finally: db.close() def test_seed_demo_scenarios_present(): db = SessionLocal() try: reset_and_seed(db) booking = db.scalar(select(Booking).where(Booking.public_ref == "BK-DEMO-RETURN")) assert booking is not None assert booking.status == "active" duplicate_customer = db.scalar(select(Customer).where(Customer.public_ref == "CUS-0178")) assert duplicate_customer is not None duplicate_issue = db.scalar( select(DataQualityIssue).where(DataQualityIssue.public_ref == "DQ-DEMO-DUPLICATE") ) assert duplicate_issue is not None assert duplicate_issue.rule_type == "possible_duplicate_customer" failed_run = db.scalar( select(OutboxEvent).where(OutboxEvent.delivery_status == "failed") ) assert failed_run is not None finally: db.close() def _by_ref(db, model, ref): return db.scalar(select(model).where(model.public_ref == ref)) def test_seed_scenario_s1_odometer_regression_return(): """S1: BK-DEMO-RETURN on MO-024 is an active booking ready for a return with a below-canonical odometer reading, using the vehicle's own current odometer.""" db = SessionLocal() try: reset_and_seed(db) booking = _by_ref(db, Booking, "BK-DEMO-RETURN") vehicle = _by_ref(db, Vehicle, "MO-024") assert booking is not None and vehicle is not None assert booking.vehicle_id == vehicle.id assert booking.status == "active" assert booking.end_odometer_km is None # A demo return reading must sit below the vehicle's canonical odometer to # reproduce the odometer-regression anomaly deterministically. assert vehicle.odometer_km > 0 finally: db.close() def test_seed_scenario_s2_duplicate_customer_pair(): """S2: CUS-0012/CUS-0178 form a possible-duplicate pair with a matching open issue.""" db = SessionLocal() try: reset_and_seed(db) primary = _by_ref(db, Customer, "CUS-0012") duplicate = _by_ref(db, Customer, "CUS-0178") assert primary is not None and duplicate is not None assert primary.email == duplicate.email assert duplicate.merged_into_customer_id is None issue = _by_ref(db, DataQualityIssue, "DQ-DEMO-DUPLICATE") assert issue is not None assert issue.rule_type == "possible_duplicate_customer" assert issue.status == "open" related = issue.evidence_json.get("related_refs", []) assert "CUS-0012" in related or "CUS-0178" in related finally: db.close() def test_seed_scenario_s4_booking_overlap(): """S4: MO-016 carries two overlapping reservations plus a matching open issue.""" db = SessionLocal() try: reset_and_seed(db) vehicle = _by_ref(db, Vehicle, "MO-016") booking_a = _by_ref(db, Booking, "BK-DEMO-OVERLAP-A") booking_b = _by_ref(db, Booking, "BK-DEMO-OVERLAP-B") assert vehicle is not None and booking_a is not None and booking_b is not None assert booking_a.vehicle_id == vehicle.id assert booking_b.vehicle_id == vehicle.id assert booking_a.starts_at < booking_b.ends_at assert booking_b.starts_at < booking_a.ends_at issue = _by_ref(db, DataQualityIssue, "DQ-DEMO-OVERLAP") assert issue is not None assert issue.rule_type == "booking_overlap" assert issue.status == "open" finally: db.close() def test_seed_scenario_s5_failed_workflow_run(): """S5: one seeded outbox event is durably 'failed' (terminal, retryable), not merely pending, so the background dispatcher never silently auto-heals it away.""" db = SessionLocal() try: reset_and_seed(db) failed = db.scalar( select(OutboxEvent).where( OutboxEvent.event_id == "00000000-0000-4000-8000-000000000020" ) ) assert failed is not None assert failed.delivery_status == "failed" assert failed.attempts >= 1 assert failed.last_error finally: db.close() def test_seed_dates_are_anchored_to_reset_moment(): """Every reset shifts seeded dates by (real today - authored anchor), so scenario bookings stay 'today'/'near-future' relative to whenever the reset actually ran, instead of decaying back to the fixed 2026-08-01 authoring date.""" db = SessionLocal() try: result = reset_and_seed(db) today = datetime.now(UTC).date() assert result.anchor_date == today shift = today - SEED_AUTHORED_ANCHOR booking = _by_ref(db, Booking, "BK-DEMO-RETURN") assert booking is not None # Authored ends_at was 2026-08-01T09:00Z; after shifting it must land on the # real reset date, not the frozen authoring date (unless shift is exactly zero). assert booking.ends_at.date() == today or shift.days == 0 marker = db.scalar( select(AuditEvent) .where(AuditEvent.action == "demo_data_seeded") .order_by(AuditEvent.occurred_at.desc()) ) assert marker is not None assert marker.metadata_json["anchor_date"] == today.isoformat() assert marker.metadata_json["seed_authored_anchor"] == SEED_AUTHORED_ANCHOR.isoformat() finally: db.close()