NuklearRabbit
a7cbeaae3b
Five rule scanners (duplicate customers, missing fields, odometer regression, booking overlap, status conflict) run automatically after seed and via an explicit scan endpoint. Issue defer/reject/merge-customers endpoints with transactional customer merge (booking rewiring, tombstone, audit). Data Quality nav + workbench UI with two-column duplicate comparison and inline (non-native) confirm. Dashboard attention items now link to issues. 35 backend tests passing, ruff clean. Fixed a real false-positive bug in odometer-regression detection found through iteration on seed data, and two TS narrowing errors. Verified end-to-end via browser: S2 merge and S4 overlap scenarios.
55 lines
2.1 KiB
Python
55 lines
2.1 KiB
Python
from sqlalchemy import func, select
|
|
|
|
from app.core.db import SessionLocal
|
|
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 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()
|