from sqlalchemy import select from app.core.db import SessionLocal from app.models.booking import Booking from app.models.vehicle import Vehicle 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 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_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_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_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_apply_recommended_status_requires_operations_manager(employee_client): response = employee_client.post( "/api/v1/data-quality/issues/DQ-DEMO-STATUS/apply-recommended-status" ) assert response.status_code == 403 def test_apply_recommended_status_resolves_conflict(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" ) assert response.status_code == 200 body = response.json() assert body["issue"]["status"] == "resolved" assert body["applied_status"] assert body["reason"] vehicle = ops_client.get(f"/api/v1/vehicles/{target['entity_ref']}").json() assert vehicle["operational_status"] == body["applied_status"] 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): # The seeded odometer_regression issues carry no related booking (CSV-only rows). # Create a fresh one with a real related booking via a live regression return, so # the "correct_reading" path has an actual booking_ref to target. 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"] 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 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 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"