383 lines
14 KiB
Python
383 lines
14 KiB
Python
import threading
|
|
from datetime import datetime, timedelta
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.main import app
|
|
|
|
|
|
def test_list_bookings_filters_by_vehicle(ops_client):
|
|
response = ops_client.get("/api/v1/bookings", params={"vehicle_ref": "MO-024"})
|
|
assert response.status_code == 200
|
|
bookings = response.json()
|
|
assert len(bookings) >= 1
|
|
assert all(b["vehicle_ref"] == "MO-024" for b in bookings)
|
|
|
|
|
|
def test_list_bookings_supports_bounded_search_pages(ops_client):
|
|
response = ops_client.get(
|
|
"/api/v1/bookings",
|
|
params={"query": "BK-", "page": 1, "page_size": 25},
|
|
)
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["page"] == 1
|
|
assert body["page_size"] == 25
|
|
assert body["total"] > 25
|
|
assert body["total_pages"] > 1
|
|
assert len(body["items"]) == 25
|
|
|
|
|
|
def test_list_bookings_filters_operational_window_and_location(ops_client):
|
|
booking = ops_client.get("/api/v1/bookings").json()[0]
|
|
vehicle = ops_client.get(f"/api/v1/vehicles/{booking['vehicle_ref']}").json()
|
|
starts_at = datetime.fromisoformat(booking["starts_at"])
|
|
response = ops_client.get(
|
|
"/api/v1/bookings",
|
|
params={
|
|
"starts_from": (starts_at - timedelta(minutes=1)).isoformat(),
|
|
"starts_to": (starts_at + timedelta(minutes=1)).isoformat(),
|
|
"location": vehicle["location"],
|
|
"sort": "starts_asc",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert booking["public_ref"] in {item["public_ref"] for item in response.json()}
|
|
|
|
|
|
def test_create_booking_rejects_overlap_and_audits_valid_booking(ops_client):
|
|
existing = ops_client.get("/api/v1/bookings/BK-DEMO-RETURN").json()
|
|
conflict = ops_client.post(
|
|
"/api/v1/bookings",
|
|
json={
|
|
"customer_ref": "CUS-0001",
|
|
"vehicle_ref": existing["vehicle_ref"],
|
|
"starts_at": existing["starts_at"],
|
|
"ends_at": existing["ends_at"],
|
|
},
|
|
)
|
|
assert conflict.status_code == 409
|
|
available_vehicle = ops_client.get("/api/v1/vehicles", params={"status": "available"}).json()[
|
|
0
|
|
]["public_ref"]
|
|
created = ops_client.post(
|
|
"/api/v1/bookings",
|
|
json={
|
|
"customer_ref": "CUS-0001",
|
|
"vehicle_ref": available_vehicle,
|
|
"starts_at": "2030-09-01T10:00:00Z",
|
|
"ends_at": "2030-09-02T12:00:00Z",
|
|
},
|
|
)
|
|
assert created.status_code == 201
|
|
body = created.json()
|
|
assert body["status"] == "reserved"
|
|
assert body["vehicle_ref"] == available_vehicle
|
|
assert body["requirements_complete"] is False
|
|
|
|
|
|
def test_booking_requirements_are_explicit_and_audited(ops_client):
|
|
window = {"starts_at": "2031-09-01T10:00:00Z", "ends_at": "2031-09-02T12:00:00Z"}
|
|
vehicle = ops_client.get("/api/v1/bookings/availability", params=window).json()[0]
|
|
booking = ops_client.post(
|
|
"/api/v1/bookings",
|
|
json={"customer_ref": "CUS-0001", "vehicle_ref": vehicle["public_ref"], **window},
|
|
).json()
|
|
checkout = ops_client.post(
|
|
f"/api/v1/bookings/{booking['public_ref']}/checkout",
|
|
json={
|
|
"start_odometer_km": 100000,
|
|
"fuel_level_percent": 90,
|
|
"cleanliness_ok": True,
|
|
"damage_reported": False,
|
|
"technical_warning": False,
|
|
},
|
|
)
|
|
assert checkout.status_code == 409
|
|
|
|
whitespace_only = ops_client.post(
|
|
f"/api/v1/bookings/{booking['public_ref']}/complete-requirements",
|
|
json={"confirmation": " "},
|
|
)
|
|
assert whitespace_only.status_code == 422
|
|
assert (
|
|
ops_client.get(f"/api/v1/bookings/{booking['public_ref']}").json()["requirements_complete"]
|
|
is False
|
|
)
|
|
|
|
confirmed = ops_client.post(
|
|
f"/api/v1/bookings/{booking['public_ref']}/complete-requirements",
|
|
json={"confirmation": "Licence and rental conditions checked"},
|
|
)
|
|
assert confirmed.status_code == 200
|
|
assert confirmed.json()["requirements_complete"] is True
|
|
audit = ops_client.get("/api/v1/audit", params={"action": "booking_requirements_completed"})
|
|
assert audit.status_code == 200
|
|
assert any(item["entity_ref"] == booking["public_ref"] for item in audit.json())
|
|
|
|
|
|
def test_booking_creation_cannot_preconfirm_requirements(ops_client):
|
|
window = {"starts_at": "2033-09-01T10:00:00Z", "ends_at": "2033-09-02T12:00:00Z"}
|
|
vehicle = ops_client.get("/api/v1/bookings/availability", params=window).json()[0]
|
|
|
|
created = ops_client.post(
|
|
"/api/v1/bookings",
|
|
json={
|
|
"customer_ref": "CUS-0001",
|
|
"vehicle_ref": vehicle["public_ref"],
|
|
"requirements_complete": True,
|
|
**window,
|
|
},
|
|
)
|
|
|
|
assert created.status_code == 201
|
|
assert created.json()["requirements_complete"] is False
|
|
audit = ops_client.get(
|
|
"/api/v1/audit",
|
|
params={"action": "booking_requirements_completed"},
|
|
)
|
|
assert all(item["entity_ref"] != created.json()["public_ref"] for item in audit.json())
|
|
|
|
|
|
def test_customer_search_returns_canonical_customers(ops_client):
|
|
response = ops_client.get("/api/v1/customers", params={"query": "CUS-"})
|
|
assert response.status_code == 200
|
|
assert response.json()
|
|
assert all(item["public_ref"].startswith("CUS-") for item in response.json())
|
|
|
|
|
|
def test_booking_availability_excludes_overlapping_vehicle(ops_client):
|
|
existing = ops_client.get("/api/v1/bookings/BK-DEMO-RETURN").json()
|
|
response = ops_client.get(
|
|
"/api/v1/bookings/availability",
|
|
params={"starts_at": existing["starts_at"], "ends_at": existing["ends_at"]},
|
|
)
|
|
assert response.status_code == 200
|
|
assert existing["vehicle_ref"] not in {item["public_ref"] for item in response.json()}
|
|
|
|
|
|
def test_reserved_booking_can_be_cancelled_once(ops_client):
|
|
window = {"starts_at": "2032-09-01T10:00:00Z", "ends_at": "2032-09-02T12:00:00Z"}
|
|
available = ops_client.get("/api/v1/bookings/availability", params=window).json()
|
|
assert available
|
|
create_response = ops_client.post(
|
|
"/api/v1/bookings",
|
|
json={
|
|
"customer_ref": "CUS-0001",
|
|
"vehicle_ref": available[0]["public_ref"],
|
|
**window,
|
|
},
|
|
)
|
|
assert create_response.status_code == 201
|
|
created = create_response.json()
|
|
response = ops_client.post(
|
|
f"/api/v1/bookings/{created['public_ref']}/cancel",
|
|
json={"reason": "Customer changed plans"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "cancelled"
|
|
repeated = ops_client.post(
|
|
f"/api/v1/bookings/{created['public_ref']}/cancel",
|
|
json={"reason": "Customer changed plans"},
|
|
)
|
|
assert repeated.status_code == 409
|
|
|
|
|
|
def test_reserved_booking_can_be_rescheduled_with_overlap_protection(ops_client):
|
|
window = {"starts_at": "2033-09-01T10:00:00Z", "ends_at": "2033-09-02T12:00:00Z"}
|
|
existing = ops_client.get("/api/v1/bookings/BK-DEMO-RETURN").json()
|
|
booking = ops_client.post(
|
|
"/api/v1/bookings",
|
|
json={"customer_ref": "CUS-0001", "vehicle_ref": existing["vehicle_ref"], **window},
|
|
).json()
|
|
updated = ops_client.patch(
|
|
f"/api/v1/bookings/{booking['public_ref']}/schedule",
|
|
json={
|
|
"starts_at": "2033-09-03T10:00:00Z",
|
|
"ends_at": "2033-09-04T12:00:00Z",
|
|
"reason": "Customer requested a later collection",
|
|
},
|
|
)
|
|
assert updated.status_code == 200
|
|
assert updated.json()["starts_at"].startswith("2033-09-03T10:00:00")
|
|
|
|
conflict = ops_client.patch(
|
|
f"/api/v1/bookings/{booking['public_ref']}/schedule",
|
|
json={
|
|
"starts_at": existing["starts_at"],
|
|
"ends_at": existing["ends_at"],
|
|
"reason": "Conflicting test move",
|
|
},
|
|
)
|
|
assert conflict.status_code == 409
|
|
|
|
|
|
def test_concurrent_bookings_only_reserve_vehicle_once():
|
|
results: list[int] = []
|
|
seed_client = TestClient(app)
|
|
seed_client.post("/api/v1/demo/login", json={"role": "operations_manager"})
|
|
window = {"starts_at": "2040-09-01T10:00:00Z", "ends_at": "2040-09-02T12:00:00Z"}
|
|
available = seed_client.get("/api/v1/bookings/availability", params=window).json()
|
|
assert available
|
|
vehicle_ref = available[0]["public_ref"]
|
|
|
|
def submit() -> None:
|
|
client = TestClient(app)
|
|
client.post("/api/v1/demo/login", json={"role": "operations_manager"})
|
|
response = client.post(
|
|
"/api/v1/bookings",
|
|
json={
|
|
"customer_ref": "CUS-0001",
|
|
"vehicle_ref": vehicle_ref,
|
|
**window,
|
|
},
|
|
)
|
|
results.append(response.status_code)
|
|
|
|
threads = [threading.Thread(target=submit) for _ in range(2)]
|
|
for thread in threads:
|
|
thread.start()
|
|
for thread in threads:
|
|
thread.join()
|
|
|
|
assert results.count(201) == 1
|
|
assert results.count(409) == 1
|
|
|
|
|
|
def test_checkout_records_inspection_and_activates_safe_booking(ops_client):
|
|
window = {"starts_at": "2050-09-01T10:00:00Z", "ends_at": "2050-09-02T12:00:00Z"}
|
|
available = ops_client.get("/api/v1/bookings/availability", params=window).json()
|
|
vehicle_option = next(item for item in available if item["operational_status"] == "available")
|
|
vehicle = ops_client.get(f"/api/v1/vehicles/{vehicle_option['public_ref']}").json()
|
|
booking = ops_client.post(
|
|
"/api/v1/bookings",
|
|
json={
|
|
"customer_ref": "CUS-0001",
|
|
"vehicle_ref": vehicle["public_ref"],
|
|
**window,
|
|
},
|
|
).json()
|
|
confirmed = ops_client.post(
|
|
f"/api/v1/bookings/{booking['public_ref']}/complete-requirements",
|
|
json={"confirmation": "Licence and rental conditions checked"},
|
|
)
|
|
assert confirmed.status_code == 200
|
|
response = ops_client.post(
|
|
f"/api/v1/bookings/{booking['public_ref']}/checkout",
|
|
json={
|
|
"start_odometer_km": vehicle["odometer_km"],
|
|
"fuel_level_percent": 95,
|
|
"cleanliness_ok": True,
|
|
"damage_reported": False,
|
|
"technical_warning": False,
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["activated"] is True
|
|
assert response.json()["booking_status"] == "active"
|
|
updated_vehicle = ops_client.get(f"/api/v1/vehicles/{vehicle['public_ref']}").json()
|
|
assert updated_vehicle["operational_status"] == "rented"
|
|
assert any(item["type"] == "checkout" for item in updated_vehicle["inspections"])
|
|
checkout_audit = ops_client.get(
|
|
"/api/v1/audit",
|
|
params={"action": "booking_checkout_recorded"},
|
|
).json()
|
|
booking_event = next(
|
|
item for item in checkout_audit if item["entity_ref"] == booking["public_ref"]
|
|
)
|
|
vehicle_audit = ops_client.get(
|
|
"/api/v1/audit",
|
|
params={
|
|
"action": "vehicle_status_changed",
|
|
"correlation_id": booking_event["correlation_id"],
|
|
},
|
|
).json()
|
|
assert len(vehicle_audit) == 1
|
|
assert vehicle_audit[0]["entity_ref"] == vehicle["public_ref"]
|
|
assert vehicle_audit[0]["before"]["operational_status"] == "available"
|
|
assert vehicle_audit[0]["after"]["operational_status"] == "rented"
|
|
|
|
|
|
def test_checkout_odometer_regression_keeps_canonical_and_opens_bounded_quality_issue(
|
|
ops_client,
|
|
):
|
|
window = {"starts_at": "2052-09-01T10:00:00Z", "ends_at": "2052-09-02T12:00:00Z"}
|
|
existing_issues = ops_client.get(
|
|
"/api/v1/data-quality/issues",
|
|
params={"status": "open", "rule_type": "odometer_regression"},
|
|
).json()
|
|
unavailable_refs = {issue["entity_ref"] for issue in existing_issues}
|
|
available = ops_client.get("/api/v1/bookings/availability", params=window).json()
|
|
vehicle_option = next(
|
|
item
|
|
for item in available
|
|
if item["operational_status"] == "available" and item["public_ref"] not in unavailable_refs
|
|
)
|
|
vehicle = ops_client.get(f"/api/v1/vehicles/{vehicle_option['public_ref']}").json()
|
|
assert vehicle["odometer_km"] > 0
|
|
booking = ops_client.post(
|
|
"/api/v1/bookings",
|
|
json={
|
|
"customer_ref": "CUS-0003",
|
|
"vehicle_ref": vehicle["public_ref"],
|
|
**window,
|
|
},
|
|
).json()
|
|
ops_client.post(
|
|
f"/api/v1/bookings/{booking['public_ref']}/complete-requirements",
|
|
json={"confirmation": "Licence and rental conditions checked"},
|
|
)
|
|
|
|
checkout = ops_client.post(
|
|
f"/api/v1/bookings/{booking['public_ref']}/checkout",
|
|
json={
|
|
"start_odometer_km": vehicle["odometer_km"] - 1,
|
|
"fuel_level_percent": 90,
|
|
"cleanliness_ok": True,
|
|
"damage_reported": False,
|
|
"technical_warning": False,
|
|
},
|
|
)
|
|
|
|
assert checkout.status_code == 200
|
|
assert checkout.json()["booking_status"] == "blocked"
|
|
updated_vehicle = ops_client.get(f"/api/v1/vehicles/{vehicle['public_ref']}").json()
|
|
assert updated_vehicle["odometer_km"] == vehicle["odometer_km"]
|
|
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"] == vehicle["public_ref"])
|
|
assert issue["evidence"]["source_type"] == "checkout"
|
|
assert issue["evidence"]["correctable_booking_refs"] == []
|
|
detail = ops_client.get(f"/api/v1/data-quality/issues/{issue['public_ref']}").json()
|
|
assert any(snapshot["entity_type"] == "inspection" for snapshot in detail["related_snapshots"])
|
|
invalid_correction = ops_client.post(
|
|
f"/api/v1/data-quality/issues/{issue['public_ref']}/resolve-odometer-regression",
|
|
json={
|
|
"decision": "correct_reading",
|
|
"booking_ref": booking["public_ref"],
|
|
"corrected_odometer_km": vehicle["odometer_km"],
|
|
},
|
|
)
|
|
assert invalid_correction.status_code == 422
|
|
retained = ops_client.post(
|
|
f"/api/v1/data-quality/issues/{issue['public_ref']}/resolve-odometer-regression",
|
|
json={"decision": "retain_canonical"},
|
|
)
|
|
assert retained.status_code == 200
|
|
assert retained.json()["status"] == "resolved"
|
|
|
|
|
|
def test_get_booking_detail(ops_client):
|
|
response = ops_client.get("/api/v1/bookings/BK-DEMO-RETURN")
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["status"] == "active"
|
|
assert body["vehicle_ref"] == "MO-024"
|
|
|
|
|
|
def test_get_booking_404_for_unknown_ref(ops_client):
|
|
response = ops_client.get("/api/v1/bookings/BK-UNKNOWN")
|
|
assert response.status_code == 404
|