1421 lines
56 KiB
Python
1421 lines
56 KiB
Python
from concurrent.futures import ThreadPoolExecutor
|
|
from datetime import UTC, datetime, timedelta
|
|
from threading import Barrier
|
|
|
|
from sqlalchemy import delete, select, text
|
|
|
|
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.idempotency import IdempotencyRecord
|
|
from app.models.inspection import Inspection
|
|
from app.models.maintenance import MaintenanceRecord
|
|
from app.models.outbox import OutboxEvent
|
|
from app.models.vehicle import Vehicle
|
|
from app.schemas import CurrentUser, RegisterReturnRequest, ResolveOdometerRegressionRequest
|
|
from app.services.data_quality import (
|
|
open_odometer_regression_issue,
|
|
resolve_odometer_regression,
|
|
)
|
|
from app.services.returns import register_vehicle_return
|
|
|
|
|
|
def _activate_booking(vehicle_ref: str, start_odometer_km: int) -> str:
|
|
db = SessionLocal()
|
|
try:
|
|
vehicle = db.scalar(select(Vehicle).where(Vehicle.public_ref == vehicle_ref))
|
|
booking = db.scalar(
|
|
select(Booking).where(Booking.vehicle_id == vehicle.id, Booking.status == "returned")
|
|
)
|
|
booking.status = "active"
|
|
booking.start_odometer_km = start_odometer_km
|
|
booking.end_odometer_km = None
|
|
db.commit()
|
|
return booking.public_ref
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def _cleanup_odometer_scenario(vehicle_id, customer_id) -> None:
|
|
"""Remove one isolated DQ-03 scenario in foreign-key-safe order."""
|
|
with SessionLocal() as db:
|
|
booking_ids = db.scalars(select(Booking.id).where(Booking.vehicle_id == vehicle_id)).all()
|
|
issue_ids = db.scalars(
|
|
select(DataQualityIssue.id).where(DataQualityIssue.entity_id == vehicle_id)
|
|
).all()
|
|
audit_entity_ids = [vehicle_id, *booking_ids, *issue_ids]
|
|
if audit_entity_ids:
|
|
db.execute(delete(AuditEvent).where(AuditEvent.entity_id.in_(audit_entity_ids)))
|
|
if booking_ids:
|
|
db.execute(
|
|
delete(IdempotencyRecord).where(IdempotencyRecord.booking_id.in_(booking_ids))
|
|
)
|
|
db.execute(delete(OutboxEvent).where(OutboxEvent.aggregate_id.in_(booking_ids)))
|
|
db.execute(delete(DataQualityIssue).where(DataQualityIssue.entity_id == vehicle_id))
|
|
db.execute(delete(Inspection).where(Inspection.vehicle_id == vehicle_id))
|
|
db.execute(delete(MaintenanceRecord).where(MaintenanceRecord.vehicle_id == vehicle_id))
|
|
db.execute(delete(Booking).where(Booking.vehicle_id == vehicle_id))
|
|
db.execute(delete(Vehicle).where(Vehicle.id == vehicle_id))
|
|
db.execute(delete(Customer).where(Customer.id == customer_id))
|
|
db.commit()
|
|
|
|
|
|
def test_list_includes_all_five_rule_types(ops_client):
|
|
response = ops_client.get("/api/v1/data-quality/issues")
|
|
assert response.status_code == 200
|
|
issues = response.json()
|
|
rule_types = {i["rule_type"] for i in issues}
|
|
assert rule_types == {
|
|
"possible_duplicate_customer",
|
|
"missing_required_field",
|
|
"odometer_regression",
|
|
"booking_overlap",
|
|
"vehicle_status_conflict",
|
|
}
|
|
|
|
|
|
def test_scan_is_idempotent_once_seeded(ops_client):
|
|
# The session fixture already ran a scan as part of seeding; running again
|
|
# must not create duplicate open issues for the same (rule_type, entity).
|
|
response = ops_client.post("/api/v1/data-quality/scan")
|
|
assert response.status_code == 200
|
|
assert response.json()["created"] == {}
|
|
|
|
|
|
def test_scan_detects_imported_maintenance_odometer_regression(ops_client):
|
|
with SessionLocal() as db:
|
|
vehicle = Vehicle(
|
|
public_ref="MO-DQ-SCAN",
|
|
make="Synthetic",
|
|
model="Scanner",
|
|
model_year=2026,
|
|
registration_number="DQ-SCAN-01",
|
|
location="Brussels",
|
|
operational_status="available",
|
|
odometer_km=20_000,
|
|
next_service_km=30_000,
|
|
active=True,
|
|
version=1,
|
|
)
|
|
db.add(vehicle)
|
|
db.flush()
|
|
vehicle_id = vehicle.id
|
|
db.add_all(
|
|
[
|
|
MaintenanceRecord(
|
|
public_ref="MAINT-DQ-SCAN-A",
|
|
vehicle_id=vehicle.id,
|
|
occurred_at=datetime(2045, 1, 1, tzinfo=UTC),
|
|
odometer_km=20_000,
|
|
category="inspection",
|
|
summary="Synthetic scan baseline",
|
|
),
|
|
MaintenanceRecord(
|
|
public_ref="MAINT-DQ-SCAN-B",
|
|
vehicle_id=vehicle.id,
|
|
occurred_at=datetime(2045, 2, 1, tzinfo=UTC),
|
|
odometer_km=19_000,
|
|
category="inspection",
|
|
summary="Synthetic imported regression",
|
|
),
|
|
]
|
|
)
|
|
db.commit()
|
|
try:
|
|
response = ops_client.post("/api/v1/data-quality/scan")
|
|
assert response.status_code == 200
|
|
assert response.json()["created"]["odometer_regression"] >= 1
|
|
issues = ops_client.get(
|
|
"/api/v1/data-quality/issues",
|
|
params={"status": "open", "rule_type": "odometer_regression"},
|
|
).json()
|
|
issue = next(item for item in issues if item["entity_ref"] == "MO-DQ-SCAN")
|
|
assert issue["evidence"]["source_type"] == "maintenance"
|
|
assert issue["evidence"]["correctable_booking_refs"] == []
|
|
|
|
# A later imported regression must enrich the existing open issue instead of
|
|
# being silently dropped by the generic check-then-return scanner path.
|
|
with SessionLocal() as db:
|
|
db.add(
|
|
MaintenanceRecord(
|
|
public_ref="MAINT-DQ-SCAN-C",
|
|
vehicle_id=vehicle_id,
|
|
occurred_at=datetime(2045, 3, 1, tzinfo=UTC),
|
|
odometer_km=18_000,
|
|
category="inspection",
|
|
summary="Second synthetic imported regression",
|
|
)
|
|
)
|
|
db.commit()
|
|
rescanned = ops_client.post("/api/v1/data-quality/scan")
|
|
assert rescanned.status_code == 200
|
|
enriched = ops_client.get(f"/api/v1/data-quality/issues/{issue['public_ref']}").json()
|
|
later_refs = {signal["params"]["later_ref"] for signal in enriched["evidence"]["signals"]}
|
|
assert {"MAINT-DQ-SCAN-B", "MAINT-DQ-SCAN-C"}.issubset(later_refs)
|
|
finally:
|
|
with SessionLocal() as db:
|
|
db.execute(delete(DataQualityIssue).where(DataQualityIssue.entity_id == vehicle_id))
|
|
db.execute(delete(MaintenanceRecord).where(MaintenanceRecord.vehicle_id == vehicle_id))
|
|
db.execute(delete(Vehicle).where(Vehicle.id == vehicle_id))
|
|
db.commit()
|
|
|
|
|
|
def test_scan_requires_operations_manager(employee_client):
|
|
response = employee_client.post("/api/v1/data-quality/scan")
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_list_issues_requires_operations_manager(employee_client):
|
|
response = employee_client.get("/api/v1/data-quality/issues")
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_manager_can_assign_prioritised_quality_work_and_filter_it(ops_client):
|
|
assignee = next(user for user in ops_client.get("/api/v1/users").json() if user["active"])
|
|
open_issues = ops_client.get("/api/v1/data-quality/issues", params={"status": "open"}).json()
|
|
refs = [issue["public_ref"] for issue in open_issues[:2]]
|
|
due_at = "2030-01-15T12:00:00+00:00"
|
|
updated = ops_client.post(
|
|
"/api/v1/data-quality/issues/bulk-work",
|
|
json={
|
|
"issue_refs": refs,
|
|
"assigned_to_ref": assignee["public_ref"],
|
|
"due_at": due_at,
|
|
},
|
|
)
|
|
assert updated.status_code == 200
|
|
assert {issue["public_ref"] for issue in updated.json()["updated"]} == set(refs)
|
|
assert all(
|
|
issue["assigned_to_ref"] == assignee["public_ref"] for issue in updated.json()["updated"]
|
|
)
|
|
|
|
filtered = ops_client.get(
|
|
"/api/v1/data-quality/issues",
|
|
params={"status": "open", "assigned_to_ref": assignee["public_ref"]},
|
|
).json()
|
|
assert set(refs).issubset({issue["public_ref"] for issue in filtered})
|
|
audits = ops_client.get("/api/v1/audit", params={"action": "data_quality_work_updated"}).json()
|
|
assert len(audits) >= 2
|
|
|
|
cleared = ops_client.post(
|
|
"/api/v1/data-quality/issues/bulk-work",
|
|
json={"issue_refs": refs, "clear_assignment": True},
|
|
)
|
|
assert cleared.status_code == 200
|
|
assert all(issue["assigned_to_ref"] is None for issue in cleared.json()["updated"])
|
|
|
|
|
|
def test_employee_cannot_assign_quality_work(employee_client):
|
|
response = employee_client.post(
|
|
"/api/v1/data-quality/issues/bulk-work",
|
|
json={"issue_refs": ["DQ-DEMO-OVERLAP"], "clear_assignment": True},
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_issue_page_preserves_severity_filter_and_limits_results(ops_client):
|
|
response = ops_client.get(
|
|
"/api/v1/data-quality/issues",
|
|
params={"severity": "high", "page": 1, "page_size": 25},
|
|
)
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert len(body["items"]) <= 25
|
|
assert all(issue["severity"] == "high" for issue in body["items"])
|
|
assert body["total"] >= len(body["items"])
|
|
|
|
|
|
def test_get_issue_requires_operations_manager(employee_client):
|
|
response = employee_client.get("/api/v1/data-quality/issues/DQ-DEMO-DUPLICATE")
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_defer_requires_operations_manager(employee_client):
|
|
response = employee_client.post("/api/v1/data-quality/issues/DQ-DEMO-OVERLAP/defer")
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_reject_requires_operations_manager(employee_client):
|
|
response = employee_client.post("/api/v1/data-quality/issues/DQ-DEMO-OVERLAP/reject")
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_s2_duplicate_customer_issue_detail(ops_client):
|
|
response = ops_client.get("/api/v1/data-quality/issues/DQ-DEMO-DUPLICATE")
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["entity_ref"] == "CUS-0012"
|
|
assert body["entity_snapshot"]["public_ref"] == "CUS-0012"
|
|
assert [s["public_ref"] for s in body["related_snapshots"]] == ["CUS-0178"]
|
|
|
|
|
|
def test_s4_booking_overlap_issue_detail(ops_client):
|
|
response = ops_client.get("/api/v1/data-quality/issues/DQ-DEMO-OVERLAP")
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["entity_ref"] == "MO-016"
|
|
assert set(body["evidence"]["related_refs"]) == {"BK-DEMO-OVERLAP-A", "BK-DEMO-OVERLAP-B"}
|
|
|
|
|
|
def test_defer_then_reject_are_rejected_on_closed_issue(ops_client):
|
|
issues = ops_client.get(
|
|
"/api/v1/data-quality/issues",
|
|
params={"rule_type": "missing_required_field", "status": "open"},
|
|
).json()
|
|
target = issues[0]["public_ref"]
|
|
|
|
deferred = ops_client.post(f"/api/v1/data-quality/issues/{target}/defer")
|
|
assert deferred.status_code == 200
|
|
assert deferred.json()["status"] == "deferred"
|
|
|
|
again = ops_client.post(f"/api/v1/data-quality/issues/{target}/reject")
|
|
assert again.status_code == 409
|
|
assert again.json()["error"]["code"] == "ISSUE_NOT_OPEN"
|
|
|
|
|
|
def test_merge_customers_requires_operations_manager(employee_client):
|
|
response = employee_client.post(
|
|
"/api/v1/data-quality/issues/DQ-DEMO-DUPLICATE/merge-customers",
|
|
json={"survivor_ref": "CUS-0012"},
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_merge_customers_rejects_unrelated_survivor(ops_client):
|
|
response = ops_client.post(
|
|
"/api/v1/data-quality/issues/DQ-DEMO-DUPLICATE/merge-customers",
|
|
json={"survivor_ref": "CUS-0099"},
|
|
)
|
|
assert response.status_code == 422
|
|
assert response.json()["error"]["code"] == "INVALID_SURVIVOR"
|
|
|
|
|
|
def test_merge_customers_s2_scenario_rewires_and_audits(ops_client):
|
|
before_bookings = ops_client.get(
|
|
"/api/v1/bookings", params={"vehicle_ref": "MO-001"}
|
|
) # warm the client session; irrelevant vehicle, just a cheap authenticated call
|
|
assert before_bookings.status_code == 200
|
|
|
|
response = ops_client.post(
|
|
"/api/v1/data-quality/issues/DQ-DEMO-DUPLICATE/merge-customers",
|
|
json={"survivor_ref": "CUS-0012", "field_overrides": {"city": "Turnhout"}},
|
|
)
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["survivor_ref"] == "CUS-0012"
|
|
assert body["loser_ref"] == "CUS-0178"
|
|
|
|
issue = ops_client.get("/api/v1/data-quality/issues/DQ-DEMO-DUPLICATE").json()
|
|
assert issue["status"] == "resolved"
|
|
|
|
audit_events = ops_client.get("/api/v1/audit", params={"action": "customer_merged"}).json()
|
|
assert len(audit_events) >= 1
|
|
|
|
# Already-resolved issue cannot be merged again.
|
|
replay = ops_client.post(
|
|
"/api/v1/data-quality/issues/DQ-DEMO-DUPLICATE/merge-customers",
|
|
json={"survivor_ref": "CUS-0012"},
|
|
)
|
|
assert replay.status_code == 409
|
|
|
|
|
|
def test_overlap_related_snapshots_are_typed_as_bookings_not_vehicles(ops_client):
|
|
body = ops_client.get("/api/v1/data-quality/issues/DQ-DEMO-OVERLAP").json()
|
|
assert len(body["related_snapshots"]) == 2
|
|
for snap in body["related_snapshots"]:
|
|
assert snap["entity_type"] == "booking"
|
|
assert snap["public_ref"] in {"BK-DEMO-OVERLAP-A", "BK-DEMO-OVERLAP-B"}
|
|
assert "starts_at" in snap and "ends_at" in snap
|
|
|
|
|
|
def _first_open(ops_client, rule_type: str) -> dict:
|
|
issues = ops_client.get(
|
|
"/api/v1/data-quality/issues", params={"rule_type": rule_type, "status": "open"}
|
|
).json()
|
|
assert issues, f"expected at least one open {rule_type} issue"
|
|
return issues[0]
|
|
|
|
|
|
def test_provide_fields_requires_operations_manager(employee_client):
|
|
response = employee_client.post(
|
|
"/api/v1/data-quality/issues/DQ-DEMO-ATTENTION/provide-fields",
|
|
json={"fields": {"registration_number": "TST-001"}},
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_provide_fields_rejects_wrong_rule_type(ops_client):
|
|
response = ops_client.post(
|
|
"/api/v1/data-quality/issues/DQ-DEMO-OVERLAP/provide-fields",
|
|
json={"fields": {"make": "Test"}},
|
|
)
|
|
assert response.status_code == 409
|
|
assert response.json()["error"]["code"] == "NOT_A_MISSING_FIELD_ISSUE"
|
|
|
|
|
|
def test_provide_fields_rejects_disallowed_field(ops_client):
|
|
target = _first_open(ops_client, "missing_required_field")
|
|
detail = ops_client.get(f"/api/v1/data-quality/issues/{target['public_ref']}").json()
|
|
disallowed = "city" if detail["entity_type"] == "customer" else "next_service_km"
|
|
response = ops_client.post(
|
|
f"/api/v1/data-quality/issues/{target['public_ref']}/provide-fields",
|
|
json={"fields": {disallowed: "anything"}},
|
|
)
|
|
assert response.status_code == 422
|
|
assert response.json()["error"]["code"] == "INVALID_FIELD"
|
|
|
|
|
|
def test_provide_fields_resolves_a_vehicle_missing_field_issue(ops_client):
|
|
issues = ops_client.get(
|
|
"/api/v1/data-quality/issues",
|
|
params={"rule_type": "missing_required_field", "status": "open"},
|
|
).json()
|
|
target = next(i for i in issues if i["entity_type"] == "vehicle")
|
|
|
|
response = ops_client.post(
|
|
f"/api/v1/data-quality/issues/{target['public_ref']}/provide-fields",
|
|
json={
|
|
"fields": {
|
|
"registration_number": "TST-999",
|
|
"make": "TestMake",
|
|
"model": "TestModel",
|
|
"location": "Depot",
|
|
}
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "resolved"
|
|
|
|
vehicle = ops_client.get(f"/api/v1/vehicles/{target['entity_ref']}").json()
|
|
assert vehicle["registration_number"] == "TST-999"
|
|
|
|
|
|
def test_vehicle_entity_snapshot_includes_registration_number(ops_client):
|
|
"""The snapshot used to omit registration_number entirely, so the 'provide missing
|
|
fields' form always showed it blank -- even for a vehicle whose plate was actually
|
|
on file, and even when a *different* field was the genuinely missing one."""
|
|
issues = ops_client.get(
|
|
"/api/v1/data-quality/issues",
|
|
params={"rule_type": "missing_required_field", "status": "open"},
|
|
).json()
|
|
target = next(i for i in issues if i["entity_type"] == "vehicle")
|
|
vehicle = ops_client.get(f"/api/v1/vehicles/{target['entity_ref']}").json()
|
|
|
|
detail = ops_client.get(f"/api/v1/data-quality/issues/{target['public_ref']}").json()
|
|
assert detail["entity_snapshot"]["registration_number"] == vehicle["registration_number"]
|
|
|
|
|
|
def test_resolve_overlap_requires_operations_manager(employee_client):
|
|
response = employee_client.post(
|
|
"/api/v1/data-quality/issues/DQ-DEMO-OVERLAP/resolve-overlap",
|
|
json={"booking_ref": "BK-DEMO-OVERLAP-A"},
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_resolve_overlap_rejects_unrelated_booking(ops_client):
|
|
response = ops_client.post(
|
|
"/api/v1/data-quality/issues/DQ-DEMO-OVERLAP/resolve-overlap",
|
|
json={"booking_ref": "BK-DEMO-RETURN"},
|
|
)
|
|
assert response.status_code == 422
|
|
assert response.json()["error"]["code"] == "INVALID_BOOKING_REFERENCE"
|
|
|
|
|
|
def test_resolve_overlap_blocks_one_booking_and_resolves(ops_client):
|
|
response = ops_client.post(
|
|
"/api/v1/data-quality/issues/DQ-DEMO-OVERLAP/resolve-overlap",
|
|
json={"booking_ref": "BK-DEMO-OVERLAP-A", "note": "Blocked the later commitment."},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "resolved"
|
|
|
|
booking = ops_client.get("/api/v1/bookings/BK-DEMO-OVERLAP-A").json()
|
|
assert booking["status"] == "blocked"
|
|
|
|
|
|
def test_status_recommendation_requires_operations_manager(employee_client):
|
|
response = employee_client.post(
|
|
"/api/v1/data-quality/issues/DQ-DEMO-STATUS/status-recommendation"
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_apply_recommended_status_requires_operations_manager(employee_client):
|
|
response = employee_client.post(
|
|
"/api/v1/data-quality/issues/DQ-DEMO-STATUS/apply-recommended-status",
|
|
json={"recommendation_token": "irrelevant"},
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_status_recommendation_preview_does_not_mutate_anything(ops_client):
|
|
target = _first_open(ops_client, "vehicle_status_conflict")
|
|
vehicle_before = ops_client.get(f"/api/v1/vehicles/{target['entity_ref']}").json()
|
|
|
|
preview_response = ops_client.post(
|
|
f"/api/v1/data-quality/issues/{target['public_ref']}/status-recommendation"
|
|
)
|
|
assert preview_response.status_code == 200
|
|
preview = preview_response.json()
|
|
assert preview["current_status"] == vehicle_before["operational_status"]
|
|
assert preview["recommendation_token"]
|
|
assert "facts" in preview
|
|
|
|
# Calling preview again (as the UI would on every open) must still not mutate.
|
|
ops_client.post(f"/api/v1/data-quality/issues/{target['public_ref']}/status-recommendation")
|
|
issue_after = ops_client.get(f"/api/v1/data-quality/issues/{target['public_ref']}").json()
|
|
vehicle_after = ops_client.get(f"/api/v1/vehicles/{target['entity_ref']}").json()
|
|
assert issue_after["status"] == "open"
|
|
assert vehicle_after["operational_status"] == vehicle_before["operational_status"]
|
|
|
|
|
|
def test_apply_recommended_status_resolves_conflict(ops_client):
|
|
targets = ops_client.get(
|
|
"/api/v1/data-quality/issues",
|
|
params={"rule_type": "vehicle_status_conflict", "status": "open"},
|
|
).json()
|
|
target = None
|
|
preview = None
|
|
for candidate in targets:
|
|
candidate_preview = ops_client.post(
|
|
f"/api/v1/data-quality/issues/{candidate['public_ref']}/status-recommendation"
|
|
).json()
|
|
if candidate_preview["safe_to_apply"]:
|
|
target = candidate
|
|
preview = candidate_preview
|
|
break
|
|
assert target is not None and preview is not None
|
|
assert preview["safe_to_apply"] is True
|
|
assert preview["manual_review_required"] is False
|
|
|
|
response = ops_client.post(
|
|
f"/api/v1/data-quality/issues/{target['public_ref']}/apply-recommended-status",
|
|
json={"recommendation_token": preview["recommendation_token"]},
|
|
)
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["issue"]["status"] == "resolved"
|
|
assert body["applied_status"] == preview["recommended_status"]
|
|
assert body["reason_code"] == preview["recommendation_code"]
|
|
|
|
vehicle = ops_client.get(f"/api/v1/vehicles/{target['entity_ref']}").json()
|
|
assert vehicle["operational_status"] == body["applied_status"]
|
|
|
|
|
|
def test_apply_recommended_status_rejects_stale_token(ops_client):
|
|
target = _first_open(ops_client, "vehicle_status_conflict")
|
|
response = ops_client.post(
|
|
f"/api/v1/data-quality/issues/{target['public_ref']}/apply-recommended-status",
|
|
json={"recommendation_token": "not-a-real-token"},
|
|
)
|
|
assert response.status_code == 409
|
|
assert response.json()["error"]["code"] == "RECOMMENDATION_STALE"
|
|
|
|
|
|
def test_resolve_odometer_regression_requires_operations_manager(employee_client):
|
|
response = employee_client.post(
|
|
"/api/v1/data-quality/issues/DQ-0007/resolve-odometer-regression",
|
|
json={"decision": "retain_canonical"},
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_resolve_odometer_regression_correction_below_canonical_is_rejected_then_retained(
|
|
ops_client,
|
|
):
|
|
target = _first_open(ops_client, "odometer_regression")
|
|
vehicle_before = ops_client.get(f"/api/v1/vehicles/{target['entity_ref']}").json()
|
|
|
|
too_low = ops_client.post(
|
|
f"/api/v1/data-quality/issues/{target['public_ref']}/resolve-odometer-regression",
|
|
json={
|
|
"decision": "correct_reading",
|
|
"booking_ref": "BK-DEMO-RETURN",
|
|
"corrected_odometer_km": max(vehicle_before["odometer_km"] - 100, 0),
|
|
},
|
|
)
|
|
assert too_low.status_code == 422
|
|
assert too_low.json()["error"]["code"] in (
|
|
"CORRECTION_BELOW_CANONICAL",
|
|
"INVALID_BOOKING_REFERENCE",
|
|
)
|
|
|
|
# The rejected attempt must not have resolved or mutated anything.
|
|
still_open = ops_client.get(f"/api/v1/data-quality/issues/{target['public_ref']}").json()
|
|
assert still_open["status"] == "open"
|
|
|
|
retained = ops_client.post(
|
|
f"/api/v1/data-quality/issues/{target['public_ref']}/resolve-odometer-regression",
|
|
json={"decision": "retain_canonical", "note": "Submitted reading treated as erroneous."},
|
|
)
|
|
assert retained.status_code == 200
|
|
assert retained.json()["status"] == "resolved"
|
|
|
|
vehicle_after = ops_client.get(f"/api/v1/vehicles/{target['entity_ref']}").json()
|
|
assert vehicle_after["odometer_km"] == vehicle_before["odometer_km"]
|
|
|
|
|
|
def test_resolve_odometer_regression_correct_reading_updates_canonical(ops_client):
|
|
# Create a live return regression to prove the complete correction chain updates
|
|
# both persisted representations of that reading (booking + return inspection).
|
|
booking_ref = _activate_booking("MO-018", start_odometer_km=12000)
|
|
vehicle_before = ops_client.get("/api/v1/vehicles/MO-018").json()
|
|
low_reading = vehicle_before["odometer_km"] - 200
|
|
returned = ops_client.post(
|
|
f"/api/v1/bookings/{booking_ref}/return",
|
|
json={
|
|
"end_odometer_km": low_reading,
|
|
"fuel_level_percent": 50,
|
|
"cleanliness_ok": True,
|
|
"damage_reported": False,
|
|
"technical_warning": False,
|
|
},
|
|
headers={"Idempotency-Key": "test-dq-odometer-correct-001"},
|
|
)
|
|
assert returned.status_code == 201
|
|
issue_ref = returned.json()["quality_issue_ref"]
|
|
inspection_ref = returned.json()["inspection_ref"]
|
|
assert issue_ref is not None
|
|
|
|
corrected = vehicle_before["odometer_km"] + 500
|
|
response = ops_client.post(
|
|
f"/api/v1/data-quality/issues/{issue_ref}/resolve-odometer-regression",
|
|
json={
|
|
"decision": "correct_reading",
|
|
"booking_ref": booking_ref,
|
|
"corrected_odometer_km": corrected,
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "resolved"
|
|
|
|
vehicle = ops_client.get("/api/v1/vehicles/MO-018").json()
|
|
assert vehicle["odometer_km"] == corrected
|
|
booking = ops_client.get(f"/api/v1/bookings/{booking_ref}").json()
|
|
assert booking["end_odometer_km"] == corrected
|
|
with SessionLocal() as db:
|
|
inspection = db.scalar(select(Inspection).where(Inspection.public_ref == inspection_ref))
|
|
assert inspection is not None
|
|
assert inspection.odometer_km == corrected
|
|
|
|
|
|
def test_correcting_one_of_two_regressions_keeps_the_other_open(ops_client):
|
|
with SessionLocal() as db:
|
|
customer = Customer(
|
|
public_ref="CUS-DQ-MULTI",
|
|
first_name="Synthetic",
|
|
last_name="Multi",
|
|
email="dq-multi@example.test",
|
|
)
|
|
vehicle = Vehicle(
|
|
public_ref="MO-DQ-MULTI",
|
|
make="Synthetic",
|
|
model="Multi",
|
|
model_year=2026,
|
|
registration_number="DQ-MULTI",
|
|
location="Brussels",
|
|
operational_status="available",
|
|
odometer_km=1_000,
|
|
next_service_km=2_000,
|
|
active=True,
|
|
version=1,
|
|
)
|
|
db.add_all([customer, vehicle])
|
|
db.flush()
|
|
bookings: list[Booking] = []
|
|
inspections: list[Inspection] = []
|
|
for index, reading in enumerate((900, 800), start=1):
|
|
booking = Booking(
|
|
public_ref=f"BK-DQ-MULTI-{index}",
|
|
customer_id=customer.id,
|
|
vehicle_id=vehicle.id,
|
|
starts_at=datetime(2046, index, 1, tzinfo=UTC),
|
|
ends_at=datetime(2046, index, 2, tzinfo=UTC),
|
|
status="returned",
|
|
start_odometer_km=reading - 10,
|
|
end_odometer_km=reading,
|
|
requirements_complete=True,
|
|
)
|
|
db.add(booking)
|
|
db.flush()
|
|
inspection = Inspection(
|
|
public_ref=f"INSP-DQ-M-{index}",
|
|
booking_id=booking.id,
|
|
vehicle_id=vehicle.id,
|
|
type="return",
|
|
fuel_level_percent=50,
|
|
cleanliness_ok=True,
|
|
damage_reported=False,
|
|
technical_warning=False,
|
|
odometer_km=reading,
|
|
completed_at=booking.ends_at,
|
|
)
|
|
db.add(inspection)
|
|
db.flush()
|
|
open_odometer_regression_issue(
|
|
db,
|
|
vehicle=vehicle,
|
|
reading_ref=inspection.public_ref,
|
|
reading_km=reading,
|
|
canonical_km=vehicle.odometer_km,
|
|
source_type="return",
|
|
related_refs=[booking.public_ref, inspection.public_ref],
|
|
correctable_booking_refs=[booking.public_ref],
|
|
public_ref="DQ-MULTI-SOURCE" if index == 1 else None,
|
|
)
|
|
bookings.append(booking)
|
|
inspections.append(inspection)
|
|
booking_refs = [booking.public_ref for booking in bookings]
|
|
vehicle_id = vehicle.id
|
|
customer_id = customer.id
|
|
db.commit()
|
|
|
|
try:
|
|
first = ops_client.post(
|
|
"/api/v1/data-quality/issues/DQ-MULTI-SOURCE/resolve-odometer-regression",
|
|
json={
|
|
"decision": "correct_reading",
|
|
"booking_ref": booking_refs[0],
|
|
"corrected_odometer_km": 1_100,
|
|
},
|
|
)
|
|
assert first.status_code == 200, first.text
|
|
assert first.json()["status"] == "open"
|
|
assert first.json()["evidence"]["correctable_booking_refs"] == [booking_refs[1]]
|
|
|
|
second = ops_client.post(
|
|
"/api/v1/data-quality/issues/DQ-MULTI-SOURCE/resolve-odometer-regression",
|
|
json={
|
|
"decision": "correct_reading",
|
|
"booking_ref": booking_refs[1],
|
|
"corrected_odometer_km": 1_200,
|
|
},
|
|
)
|
|
assert second.status_code == 200, second.text
|
|
assert second.json()["status"] == "resolved"
|
|
with SessionLocal() as db:
|
|
persisted = db.scalars(
|
|
select(Inspection)
|
|
.where(Inspection.vehicle_id == vehicle_id)
|
|
.order_by(Inspection.public_ref)
|
|
).all()
|
|
assert [item.odometer_km for item in persisted] == [1_100, 1_200]
|
|
finally:
|
|
with SessionLocal() as db:
|
|
db.execute(delete(DataQualityIssue).where(DataQualityIssue.entity_id == vehicle_id))
|
|
db.execute(delete(Inspection).where(Inspection.vehicle_id == vehicle_id))
|
|
db.execute(delete(Booking).where(Booking.vehicle_id == vehicle_id))
|
|
db.execute(delete(Vehicle).where(Vehicle.id == vehicle_id))
|
|
db.execute(delete(Customer).where(Customer.id == customer_id))
|
|
db.commit()
|
|
|
|
|
|
def test_retained_regression_is_suppressed_until_source_reading_changes(ops_client):
|
|
with SessionLocal() as db:
|
|
vehicle = Vehicle(
|
|
public_ref="MO-DQ-RETAIN",
|
|
make="Synthetic",
|
|
model="Retain",
|
|
model_year=2026,
|
|
registration_number="DQ-RETAIN",
|
|
location="Brussels",
|
|
operational_status="available",
|
|
odometer_km=20_000,
|
|
next_service_km=30_000,
|
|
active=True,
|
|
version=1,
|
|
)
|
|
db.add(vehicle)
|
|
db.flush()
|
|
vehicle_id = vehicle.id
|
|
db.add_all(
|
|
[
|
|
MaintenanceRecord(
|
|
public_ref="MAINT-DQ-RET-A",
|
|
vehicle_id=vehicle.id,
|
|
occurred_at=datetime(2047, 1, 1, tzinfo=UTC),
|
|
odometer_km=20_000,
|
|
category="inspection",
|
|
summary="Retain baseline",
|
|
),
|
|
MaintenanceRecord(
|
|
public_ref="MAINT-DQ-RET-B",
|
|
vehicle_id=vehicle.id,
|
|
occurred_at=datetime(2047, 2, 1, tzinfo=UTC),
|
|
odometer_km=19_000,
|
|
category="inspection",
|
|
summary="Retained source reading",
|
|
),
|
|
]
|
|
)
|
|
db.commit()
|
|
try:
|
|
assert ops_client.post("/api/v1/data-quality/scan").status_code == 200
|
|
issue = next(
|
|
item
|
|
for item in ops_client.get(
|
|
"/api/v1/data-quality/issues",
|
|
params={"status": "open", "rule_type": "odometer_regression"},
|
|
).json()
|
|
if item["entity_ref"] == "MO-DQ-RETAIN"
|
|
)
|
|
retained = ops_client.post(
|
|
f"/api/v1/data-quality/issues/{issue['public_ref']}/resolve-odometer-regression",
|
|
json={"decision": "retain_canonical", "note": "Verified source entry"},
|
|
)
|
|
assert retained.status_code == 200
|
|
assert retained.json()["status"] == "resolved"
|
|
|
|
assert ops_client.post("/api/v1/data-quality/scan").status_code == 200
|
|
with SessionLocal() as db:
|
|
assert (
|
|
db.scalar(
|
|
select(DataQualityIssue).where(
|
|
DataQualityIssue.entity_id == vehicle_id,
|
|
DataQualityIssue.status == "open",
|
|
)
|
|
)
|
|
is None
|
|
)
|
|
changed = db.scalar(
|
|
select(MaintenanceRecord).where(MaintenanceRecord.public_ref == "MAINT-DQ-RET-B")
|
|
)
|
|
changed.odometer_km = 18_999
|
|
db.commit()
|
|
|
|
assert ops_client.post("/api/v1/data-quality/scan").status_code == 200
|
|
reopened = next(
|
|
item
|
|
for item in ops_client.get(
|
|
"/api/v1/data-quality/issues",
|
|
params={"status": "open", "rule_type": "odometer_regression"},
|
|
).json()
|
|
if item["entity_ref"] == "MO-DQ-RETAIN"
|
|
)
|
|
assert reopened["evidence"]["reopened_from"] == issue["public_ref"]
|
|
finally:
|
|
with SessionLocal() as db:
|
|
db.execute(delete(DataQualityIssue).where(DataQualityIssue.entity_id == vehicle_id))
|
|
db.execute(delete(MaintenanceRecord).where(MaintenanceRecord.vehicle_id == vehicle_id))
|
|
db.execute(delete(Vehicle).where(Vehicle.id == vehicle_id))
|
|
db.commit()
|
|
|
|
|
|
def test_live_return_retained_as_canonical_is_not_reopened_by_scan(ops_client):
|
|
now = datetime.now(UTC)
|
|
with SessionLocal() as db:
|
|
customer = Customer(
|
|
public_ref="CUS-DQ-LIVE-RET",
|
|
first_name="Synthetic",
|
|
last_name="Retained return",
|
|
email="dq-live-retain@example.test",
|
|
)
|
|
vehicle = Vehicle(
|
|
public_ref="MO-DQ-LIVE-RET",
|
|
make="Synthetic",
|
|
model="Retained return",
|
|
model_year=2026,
|
|
registration_number="DQ-LIVE-RET",
|
|
location="Brussels",
|
|
operational_status="rented",
|
|
odometer_km=20_000,
|
|
next_service_km=30_000,
|
|
active=True,
|
|
version=1,
|
|
)
|
|
db.add_all([customer, vehicle])
|
|
db.flush()
|
|
booking = Booking(
|
|
public_ref="BK-DQ-LIVE-RET",
|
|
customer_id=customer.id,
|
|
vehicle_id=vehicle.id,
|
|
starts_at=now - timedelta(days=2),
|
|
ends_at=now + timedelta(days=1),
|
|
status="active",
|
|
start_odometer_km=19_500,
|
|
end_odometer_km=None,
|
|
requirements_complete=True,
|
|
)
|
|
db.add_all(
|
|
[
|
|
booking,
|
|
MaintenanceRecord(
|
|
public_ref="MNT-DQ-LIVE-RET",
|
|
vehicle_id=vehicle.id,
|
|
occurred_at=now - timedelta(days=3),
|
|
odometer_km=20_000,
|
|
category="inspection",
|
|
summary="Synthetic canonical baseline",
|
|
),
|
|
]
|
|
)
|
|
vehicle_id = vehicle.id
|
|
customer_id = customer.id
|
|
db.commit()
|
|
|
|
try:
|
|
returned = ops_client.post(
|
|
"/api/v1/bookings/BK-DQ-LIVE-RET/return",
|
|
json={
|
|
"end_odometer_km": 19_000,
|
|
"fuel_level_percent": 50,
|
|
"cleanliness_ok": True,
|
|
"damage_reported": False,
|
|
"technical_warning": False,
|
|
},
|
|
headers={"Idempotency-Key": "test-dq-live-retain-001"},
|
|
)
|
|
assert returned.status_code == 201, returned.text
|
|
issue_ref = returned.json()["quality_issue_ref"]
|
|
inspection_ref = returned.json()["inspection_ref"]
|
|
assert issue_ref is not None
|
|
|
|
retained = ops_client.post(
|
|
f"/api/v1/data-quality/issues/{issue_ref}/resolve-odometer-regression",
|
|
json={"decision": "retain_canonical", "note": "Source reading verified as wrong."},
|
|
)
|
|
assert retained.status_code == 200, retained.text
|
|
assert retained.json()["status"] == "resolved"
|
|
assert retained.json()["evidence"]["retained_odometer_fingerprints"] == [
|
|
{
|
|
"source_type": "return",
|
|
"later_ref": inspection_ref,
|
|
"later_km": 19_000,
|
|
}
|
|
]
|
|
|
|
scanned = ops_client.post("/api/v1/data-quality/scan")
|
|
assert scanned.status_code == 200, scanned.text
|
|
with SessionLocal() as db:
|
|
issues = db.scalars(
|
|
select(DataQualityIssue)
|
|
.where(
|
|
DataQualityIssue.entity_id == vehicle_id,
|
|
DataQualityIssue.rule_type == "odometer_regression",
|
|
)
|
|
.order_by(DataQualityIssue.detected_at)
|
|
).all()
|
|
assert [(issue.public_ref, issue.status) for issue in issues] == [
|
|
(issue_ref, "resolved")
|
|
]
|
|
finally:
|
|
_cleanup_odometer_scenario(vehicle_id, customer_id)
|
|
|
|
|
|
def test_scanned_return_regression_can_correct_booking_all_inspections_and_vehicle(ops_client):
|
|
now = datetime.now(UTC)
|
|
with SessionLocal() as db:
|
|
customer = Customer(
|
|
public_ref="CUS-DQ-SCAN-RET",
|
|
first_name="Synthetic",
|
|
last_name="Scanned return",
|
|
email="dq-scan-return@example.test",
|
|
)
|
|
vehicle = Vehicle(
|
|
public_ref="MO-DQ-SCAN-RET",
|
|
make="Synthetic",
|
|
model="Scanned return",
|
|
model_year=2026,
|
|
registration_number="DQ-SCAN-RET",
|
|
location="Brussels",
|
|
operational_status="available",
|
|
odometer_km=20_000,
|
|
next_service_km=30_000,
|
|
active=True,
|
|
version=1,
|
|
)
|
|
db.add_all([customer, vehicle])
|
|
db.flush()
|
|
booking = Booking(
|
|
public_ref="BK-DQ-SCAN-RET",
|
|
customer_id=customer.id,
|
|
vehicle_id=vehicle.id,
|
|
starts_at=now - timedelta(days=3),
|
|
ends_at=now - timedelta(hours=12),
|
|
status="returned",
|
|
start_odometer_km=19_500,
|
|
end_odometer_km=18_500,
|
|
requirements_complete=True,
|
|
)
|
|
db.add(booking)
|
|
db.flush()
|
|
db.add_all(
|
|
[
|
|
MaintenanceRecord(
|
|
public_ref="MNT-DQ-SCAN-RET",
|
|
vehicle_id=vehicle.id,
|
|
occurred_at=now - timedelta(days=4),
|
|
odometer_km=20_000,
|
|
category="inspection",
|
|
summary="Synthetic canonical baseline",
|
|
),
|
|
Inspection(
|
|
public_ref="INSP-DQ-SCAN-R1",
|
|
booking_id=booking.id,
|
|
vehicle_id=vehicle.id,
|
|
type="return",
|
|
fuel_level_percent=50,
|
|
cleanliness_ok=True,
|
|
damage_reported=False,
|
|
technical_warning=False,
|
|
odometer_km=19_000,
|
|
completed_at=now - timedelta(days=1),
|
|
),
|
|
Inspection(
|
|
public_ref="INSP-DQ-SCAN-R2",
|
|
booking_id=booking.id,
|
|
vehicle_id=vehicle.id,
|
|
type="return",
|
|
fuel_level_percent=50,
|
|
cleanliness_ok=True,
|
|
damage_reported=False,
|
|
technical_warning=False,
|
|
odometer_km=18_500,
|
|
completed_at=now - timedelta(hours=12),
|
|
),
|
|
]
|
|
)
|
|
vehicle_id = vehicle.id
|
|
customer_id = customer.id
|
|
db.commit()
|
|
|
|
try:
|
|
scanned = ops_client.post("/api/v1/data-quality/scan")
|
|
assert scanned.status_code == 200, scanned.text
|
|
issue = next(
|
|
item
|
|
for item in ops_client.get(
|
|
"/api/v1/data-quality/issues",
|
|
params={"status": "open", "rule_type": "odometer_regression"},
|
|
).json()
|
|
if item["entity_ref"] == "MO-DQ-SCAN-RET"
|
|
)
|
|
assert issue["evidence"]["correctable_booking_refs"] == ["BK-DQ-SCAN-RET"]
|
|
later_refs = {signal["params"]["later_ref"] for signal in issue["evidence"]["signals"]}
|
|
assert later_refs == {"INSP-DQ-SCAN-R1", "INSP-DQ-SCAN-R2"}
|
|
|
|
corrected = ops_client.post(
|
|
f"/api/v1/data-quality/issues/{issue['public_ref']}/resolve-odometer-regression",
|
|
json={
|
|
"decision": "correct_reading",
|
|
"booking_ref": "BK-DQ-SCAN-RET",
|
|
"corrected_odometer_km": 20_500,
|
|
},
|
|
)
|
|
assert corrected.status_code == 200, corrected.text
|
|
assert corrected.json()["status"] == "resolved"
|
|
|
|
with SessionLocal() as db:
|
|
persisted_vehicle = db.get(Vehicle, vehicle_id)
|
|
persisted_booking = db.scalar(
|
|
select(Booking).where(Booking.public_ref == "BK-DQ-SCAN-RET")
|
|
)
|
|
assert persisted_booking is not None
|
|
persisted_inspections = db.scalars(
|
|
select(Inspection)
|
|
.where(Inspection.booking_id == persisted_booking.id, Inspection.type == "return")
|
|
.order_by(Inspection.public_ref)
|
|
).all()
|
|
assert persisted_vehicle is not None
|
|
assert persisted_vehicle.odometer_km == 20_500
|
|
assert persisted_booking.end_odometer_km == 20_500
|
|
assert [inspection.odometer_km for inspection in persisted_inspections] == [
|
|
20_500,
|
|
20_500,
|
|
]
|
|
finally:
|
|
_cleanup_odometer_scenario(vehicle_id, customer_id)
|
|
|
|
|
|
def test_resolver_and_concurrent_return_finish_without_deadlock_or_lost_evidence():
|
|
now = datetime.now(UTC)
|
|
with SessionLocal() as db:
|
|
customer = Customer(
|
|
public_ref="CUS-DQ-RACE",
|
|
first_name="Synthetic",
|
|
last_name="Race",
|
|
email="dq-race@example.test",
|
|
)
|
|
vehicle = Vehicle(
|
|
public_ref="MO-DQ-RACE",
|
|
make="Synthetic",
|
|
model="Race",
|
|
model_year=2026,
|
|
registration_number="DQ-RACE",
|
|
location="Brussels",
|
|
operational_status="rented",
|
|
odometer_km=20_000,
|
|
next_service_km=30_000,
|
|
active=True,
|
|
version=1,
|
|
)
|
|
db.add_all([customer, vehicle])
|
|
db.flush()
|
|
corrected_booking = Booking(
|
|
public_ref="BK-DQ-RACE-OLD",
|
|
customer_id=customer.id,
|
|
vehicle_id=vehicle.id,
|
|
starts_at=now - timedelta(days=4),
|
|
ends_at=now - timedelta(days=3),
|
|
status="returned",
|
|
start_odometer_km=19_500,
|
|
end_odometer_km=19_000,
|
|
requirements_complete=True,
|
|
)
|
|
concurrent_booking = Booking(
|
|
public_ref="BK-DQ-RACE-NEW",
|
|
customer_id=customer.id,
|
|
vehicle_id=vehicle.id,
|
|
starts_at=now - timedelta(days=1),
|
|
ends_at=now + timedelta(days=1),
|
|
status="active",
|
|
start_odometer_km=19_000,
|
|
end_odometer_km=None,
|
|
requirements_complete=True,
|
|
)
|
|
db.add_all([corrected_booking, concurrent_booking])
|
|
db.flush()
|
|
old_inspection = Inspection(
|
|
public_ref="INSP-DQ-RACE-OLD",
|
|
booking_id=corrected_booking.id,
|
|
vehicle_id=vehicle.id,
|
|
type="return",
|
|
fuel_level_percent=50,
|
|
cleanliness_ok=True,
|
|
damage_reported=False,
|
|
technical_warning=False,
|
|
odometer_km=19_000,
|
|
completed_at=corrected_booking.ends_at,
|
|
)
|
|
db.add(old_inspection)
|
|
db.flush()
|
|
issue = open_odometer_regression_issue(
|
|
db,
|
|
vehicle=vehicle,
|
|
reading_ref=old_inspection.public_ref,
|
|
reading_km=19_000,
|
|
canonical_km=20_000,
|
|
source_type="return",
|
|
related_refs=[corrected_booking.public_ref, old_inspection.public_ref],
|
|
correctable_booking_refs=[corrected_booking.public_ref],
|
|
public_ref="DQ-ODO-RACE",
|
|
)
|
|
assert issue is not None
|
|
vehicle_id = vehicle.id
|
|
customer_id = customer.id
|
|
db.commit()
|
|
|
|
actor = CurrentUser(
|
|
public_ref="USR-DQ-RACE",
|
|
display_name="DQ Race Manager",
|
|
role="operations_manager",
|
|
)
|
|
start = Barrier(2)
|
|
|
|
def correct_existing_reading() -> str:
|
|
with SessionLocal() as db:
|
|
db.execute(text("SET LOCAL lock_timeout = '5s'"))
|
|
start.wait(timeout=5)
|
|
resolved = resolve_odometer_regression(
|
|
db,
|
|
"DQ-ODO-RACE",
|
|
ResolveOdometerRegressionRequest(
|
|
decision="correct_reading",
|
|
booking_ref="BK-DQ-RACE-OLD",
|
|
corrected_odometer_km=20_500,
|
|
),
|
|
actor,
|
|
)
|
|
return resolved.status
|
|
|
|
def return_other_booking() -> dict:
|
|
with SessionLocal() as db:
|
|
db.execute(text("SET LOCAL lock_timeout = '5s'"))
|
|
start.wait(timeout=5)
|
|
_status, response = register_vehicle_return(
|
|
db,
|
|
"BK-DQ-RACE-NEW",
|
|
RegisterReturnRequest(
|
|
end_odometer_km=18_000,
|
|
fuel_level_percent=50,
|
|
cleanliness_ok=True,
|
|
damage_reported=False,
|
|
technical_warning=False,
|
|
),
|
|
"test-dq-resolve-return-race-001",
|
|
actor,
|
|
)
|
|
return response
|
|
|
|
try:
|
|
with ThreadPoolExecutor(max_workers=2) as executor:
|
|
correction_future = executor.submit(correct_existing_reading)
|
|
return_future = executor.submit(return_other_booking)
|
|
correction_status = correction_future.result(timeout=15)
|
|
return_result = return_future.result(timeout=15)
|
|
|
|
assert correction_status in {"open", "resolved"}
|
|
assert return_result["quality_issue_ref"] is not None
|
|
with SessionLocal() as db:
|
|
persisted_vehicle = db.get(Vehicle, vehicle_id)
|
|
bookings = {
|
|
booking.public_ref: booking
|
|
for booking in db.scalars(
|
|
select(Booking).where(Booking.vehicle_id == vehicle_id)
|
|
).all()
|
|
}
|
|
old_inspections = db.scalars(
|
|
select(Inspection).where(Inspection.booking_id == bookings["BK-DQ-RACE-OLD"].id)
|
|
).all()
|
|
open_issues = db.scalars(
|
|
select(DataQualityIssue).where(
|
|
DataQualityIssue.entity_id == vehicle_id,
|
|
DataQualityIssue.rule_type == "odometer_regression",
|
|
DataQualityIssue.status == "open",
|
|
)
|
|
).all()
|
|
assert persisted_vehicle is not None
|
|
assert persisted_vehicle.odometer_km == 20_500
|
|
assert bookings["BK-DQ-RACE-OLD"].end_odometer_km == 20_500
|
|
assert [inspection.odometer_km for inspection in old_inspections] == [20_500]
|
|
assert bookings["BK-DQ-RACE-NEW"].status == "returned"
|
|
assert bookings["BK-DQ-RACE-NEW"].end_odometer_km == 18_000
|
|
assert len(open_issues) == 1
|
|
assert open_issues[0].evidence_json["correctable_booking_refs"] == ["BK-DQ-RACE-NEW"]
|
|
later_refs = {
|
|
signal["params"]["later_ref"] for signal in open_issues[0].evidence_json["signals"]
|
|
}
|
|
assert later_refs == {return_result["inspection_ref"]}
|
|
finally:
|
|
_cleanup_odometer_scenario(vehicle_id, customer_id)
|
|
|
|
|
|
def test_early_return_uses_inspection_time_without_duplicate_booking_regression(ops_client):
|
|
with SessionLocal() as db:
|
|
customer = Customer(
|
|
public_ref="CUS-DQ-EARLY",
|
|
first_name="Synthetic",
|
|
last_name="Early",
|
|
email="dq-early@example.test",
|
|
)
|
|
vehicle = Vehicle(
|
|
public_ref="MO-DQ-EARLY",
|
|
make="Synthetic",
|
|
model="Early",
|
|
model_year=2026,
|
|
registration_number="DQ-EARLY",
|
|
location="Brussels",
|
|
operational_status="available",
|
|
odometer_km=200,
|
|
next_service_km=10_000,
|
|
active=True,
|
|
version=1,
|
|
)
|
|
db.add_all([customer, vehicle])
|
|
db.flush()
|
|
early = Booking(
|
|
public_ref="BK-DQ-EARLY-A",
|
|
customer_id=customer.id,
|
|
vehicle_id=vehicle.id,
|
|
starts_at=datetime(2048, 1, 1, tzinfo=UTC),
|
|
ends_at=datetime(2048, 3, 1, tzinfo=UTC),
|
|
status="returned",
|
|
start_odometer_km=90,
|
|
end_odometer_km=100,
|
|
requirements_complete=True,
|
|
)
|
|
later = Booking(
|
|
public_ref="BK-DQ-EARLY-B",
|
|
customer_id=customer.id,
|
|
vehicle_id=vehicle.id,
|
|
starts_at=datetime(2048, 1, 10, tzinfo=UTC),
|
|
ends_at=datetime(2048, 2, 1, tzinfo=UTC),
|
|
status="returned",
|
|
start_odometer_km=100,
|
|
end_odometer_km=200,
|
|
requirements_complete=True,
|
|
)
|
|
db.add_all([early, later])
|
|
db.flush()
|
|
db.add_all(
|
|
[
|
|
Inspection(
|
|
public_ref="INSP-DQ-EARLY-A",
|
|
booking_id=early.id,
|
|
vehicle_id=vehicle.id,
|
|
type="return",
|
|
fuel_level_percent=50,
|
|
cleanliness_ok=True,
|
|
damage_reported=False,
|
|
technical_warning=False,
|
|
odometer_km=100,
|
|
completed_at=datetime(2048, 1, 2, tzinfo=UTC),
|
|
),
|
|
Inspection(
|
|
public_ref="INSP-DQ-EARLY-B",
|
|
booking_id=later.id,
|
|
vehicle_id=vehicle.id,
|
|
type="return",
|
|
fuel_level_percent=50,
|
|
cleanliness_ok=True,
|
|
damage_reported=False,
|
|
technical_warning=False,
|
|
odometer_km=200,
|
|
completed_at=datetime(2048, 2, 1, tzinfo=UTC),
|
|
),
|
|
]
|
|
)
|
|
vehicle_id = vehicle.id
|
|
customer_id = customer.id
|
|
db.commit()
|
|
try:
|
|
assert ops_client.post("/api/v1/data-quality/scan").status_code == 200
|
|
with SessionLocal() as db:
|
|
assert (
|
|
db.scalar(
|
|
select(DataQualityIssue).where(
|
|
DataQualityIssue.entity_id == vehicle_id,
|
|
DataQualityIssue.rule_type == "odometer_regression",
|
|
)
|
|
)
|
|
is None
|
|
)
|
|
finally:
|
|
with SessionLocal() as db:
|
|
db.execute(delete(DataQualityIssue).where(DataQualityIssue.entity_id == vehicle_id))
|
|
db.execute(delete(Inspection).where(Inspection.vehicle_id == vehicle_id))
|
|
db.execute(delete(Booking).where(Booking.vehicle_id == vehicle_id))
|
|
db.execute(delete(Vehicle).where(Vehicle.id == vehicle_id))
|
|
db.execute(delete(Customer).where(Customer.id == customer_id))
|
|
db.commit()
|
|
|
|
|
|
def test_manual_scan_records_audit_event(ops_client):
|
|
scan = ops_client.post("/api/v1/data-quality/scan")
|
|
assert scan.status_code == 200
|
|
|
|
events = ops_client.get("/api/v1/audit", params={"action": "data_quality_scan_run"}).json()
|
|
assert len(events) >= 1
|
|
assert "created" in events[0]["metadata"]
|
|
|
|
|
|
def _reset_demo(ops_client) -> None:
|
|
# /api/v1/demo/reset deletes the session cookie (the reset recreates the users
|
|
# table, so the old session's user id no longer exists) -- the caller must log back
|
|
# in before making any further authenticated call with the same client.
|
|
response = ops_client.post("/api/v1/demo/reset")
|
|
assert response.status_code == 200, response.text
|
|
login_response = ops_client.post("/api/v1/demo/login", json={"role": "operations_manager"})
|
|
assert login_response.status_code == 200, login_response.text
|
|
|
|
|
|
def _resolve_overlap_issue(ops_client, *, booking_to_block: str) -> None:
|
|
overlap = _first_open(ops_client, "booking_overlap")
|
|
response = ops_client.post(
|
|
f"/api/v1/data-quality/issues/{overlap['public_ref']}/resolve-overlap",
|
|
json={"booking_ref": booking_to_block},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
assert response.json()["status"] == "resolved"
|
|
|
|
|
|
def _first_open_for_vehicle(ops_client, rule_type: str, vehicle_ref: str) -> dict:
|
|
issues = ops_client.get(
|
|
"/api/v1/data-quality/issues", params={"rule_type": rule_type, "status": "open"}
|
|
).json()
|
|
match = next((i for i in issues if i["entity_ref"] == vehicle_ref), None)
|
|
assert match, f"expected an open {rule_type} issue for {vehicle_ref}"
|
|
return match
|
|
|
|
|
|
def _apply_status_recommendation(ops_client, public_ref: str) -> dict:
|
|
preview = ops_client.post(
|
|
f"/api/v1/data-quality/issues/{public_ref}/status-recommendation"
|
|
).json()
|
|
apply_response = ops_client.post(
|
|
f"/api/v1/data-quality/issues/{public_ref}/apply-recommended-status",
|
|
json={"recommendation_token": preview["recommendation_token"]},
|
|
)
|
|
assert apply_response.status_code == 200, apply_response.text
|
|
return apply_response.json()
|
|
|
|
|
|
def test_mo_016_status_conflict_recommendation_is_order_independent(ops_client):
|
|
# MO-016 carries both a booking_overlap (DQ-DEMO-OVERLAP) and a vehicle_status_conflict
|
|
# (DQ-DEMO-STATUS) issue at once. Order independence does NOT mean "the same final
|
|
# vehicle status regardless of order" -- resolving the overlap first genuinely removes
|
|
# the conflict, so there is correctly nothing left to apply. What must hold in either
|
|
# order: the recommendation always reflects the real, current facts (never a stale
|
|
# "was some other issue open" proxy), and nothing unsafe is ever applied (never
|
|
# "rented", never a status change once the underlying condition has already resolved
|
|
# itself). See docs/fleet-ops-correction/current-gap-audit.md §6-7 and
|
|
# vehicle-status-decision-table.md.
|
|
|
|
# Order A: resolve the booking overlap first. The status-conflict issue's own
|
|
# recommendation must now correctly report that the conflict is gone -- nothing unsafe
|
|
# should be auto-applied, and the vehicle (never touched) stays exactly as it was.
|
|
_reset_demo(ops_client)
|
|
_resolve_overlap_issue(ops_client, booking_to_block="BK-DEMO-OVERLAP-B")
|
|
status_issue_a = _first_open_for_vehicle(ops_client, "vehicle_status_conflict", "MO-016")
|
|
preview_a = ops_client.post(
|
|
f"/api/v1/data-quality/issues/{status_issue_a['public_ref']}/status-recommendation"
|
|
).json()
|
|
assert preview_a["recommendation_code"] == "vehicle.no_conflict"
|
|
assert preview_a["recommended_status"] is None
|
|
assert preview_a["safe_to_apply"] is False
|
|
vehicle_a = ops_client.get("/api/v1/vehicles/MO-016").json()
|
|
assert vehicle_a["operational_status"] == "available"
|
|
|
|
# Order B: resolve the status conflict first, while the overlap is still open -- the
|
|
# conflict genuinely still exists, so the evaluator must still detect it and safely
|
|
# resolve it (never "rented").
|
|
_reset_demo(ops_client)
|
|
status_issue_b = _first_open_for_vehicle(ops_client, "vehicle_status_conflict", "MO-016")
|
|
result_b = _apply_status_recommendation(ops_client, status_issue_b["public_ref"])
|
|
assert result_b["applied_status"] != "rented"
|
|
vehicle_b_mid = ops_client.get("/api/v1/vehicles/MO-016").json()
|
|
assert vehicle_b_mid["operational_status"] == result_b["applied_status"]
|
|
|
|
# Resolving the now-redundant overlap afterwards must not itself change the vehicle's
|
|
# status as a side effect.
|
|
_resolve_overlap_issue(ops_client, booking_to_block="BK-DEMO-OVERLAP-B")
|
|
vehicle_b = ops_client.get("/api/v1/vehicles/MO-016").json()
|
|
assert vehicle_b["operational_status"] == result_b["applied_status"]
|
|
assert vehicle_b["operational_status"] != "rented"
|
|
|
|
_reset_demo(ops_client)
|
|
|
|
|
|
def test_rejected_issue_recurrence_links_to_prior_decision(ops_client):
|
|
# Reject an open vehicle_status_conflict issue without changing the vehicle, so the
|
|
# next scan re-detects the same unresolved condition -- it must not silently vanish
|
|
# or reopen the old row, but the new issue should stay linked to the rejection.
|
|
target = _first_open(ops_client, "vehicle_status_conflict")
|
|
rejected = ops_client.post(f"/api/v1/data-quality/issues/{target['public_ref']}/reject")
|
|
assert rejected.status_code == 200
|
|
|
|
rescan = ops_client.post("/api/v1/data-quality/scan")
|
|
assert rescan.status_code == 200
|
|
assert rescan.json()["created"].get("vehicle_status_conflict", 0) >= 1
|
|
|
|
reopened = ops_client.get(
|
|
"/api/v1/data-quality/issues",
|
|
params={"rule_type": "vehicle_status_conflict", "status": "open"},
|
|
).json()
|
|
match = next(
|
|
(i for i in reopened if i["evidence"].get("reopened_from") == target["public_ref"]), None
|
|
)
|
|
assert match is not None, "expected a new issue linked back to the rejected one"
|
|
assert match["evidence"]["previous_decision"] == "rejected"
|
|
|
|
|
|
def test_scan_public_refs_are_collision_resistant() -> None:
|
|
from app.services.data_quality import _new_scan_ref
|
|
|
|
refs = {_new_scan_ref("DQ-SCAN") for _ in range(1000)}
|
|
assert len(refs) == 1000
|
|
assert all(ref.startswith("DQ-SCAN-") and len(ref) == 18 for ref in refs)
|