MobilityOps/backend/tests/test_data_quality.py

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_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):
    target = _first_open(ops_client, "vehicle_status_conflict")
    preview = ops_client.post(
        f"/api/v1/data-quality/issues/{target['public_ref']}/status-recommendation"
    ).json()
    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):
    # 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 _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"