NuklearRabbit
0091c57c7f
Transactional return command with idempotency, row-lock concurrency control, odometer-regression handling, vehicle status derivation, outbox event, audit trail. Result-summary UI on booking detail. 26 backend tests passing, ruff clean. Verified end-to-end via browser against S1 demo scenario; fixed two real defects found only through browser testing (UI state loss on status transition, unflushed UUID default).
54 lines
2.0 KiB
Python
54 lines
2.0 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
|
|
assert db.scalar(select(func.count()).select_from(DataQualityIssue)) == 15
|
|
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()
|