feat(search): add role-aware backend search and truthful n8n status

Two new endpoints. GET /api/v1/search returns bounded typed results
(vehicle, booking, data-quality-issue, application section) instead of the
frontend guessing routes from regex patterns against public-ref prefixes;
data-quality and manager-only sections are filtered server-side by role,
and customers are deliberately never returned since no customer detail
route exists in this PoC.

GET /api/v1/integrations/status aggregates outbox delivery counts
(pending/delivering/succeeded/failed) into a single truthful n8n state
(disabled/unavailable/degraded/operational/no_evidence) instead of the UI
showing whichever status the single most recent event happened to be in --
a vehicle_status_conflict-style bug where one stale failure or one lucky
success could misreport the dispatcher's actual health.

Also fixes a real config gap this surfaced: MCP_HUB_REGISTRATION_ENABLED
was documented in .env.example but had no corresponding Settings field, so
it was silently ignored by pydantic-settings' extra="ignore" and never
actually read anywhere in the codebase.
This commit is contained in:
NuklearRabbit
2026-08-02 06:41:56 +02:00
parent 4bc3e33953
commit 4437b8792a
11 changed files with 596 additions and 46 deletions
+57
View File
@@ -0,0 +1,57 @@
def test_search_requires_authentication(client):
response = client.get("/api/v1/search", params={"q": "MO-001"})
assert response.status_code == 401
def test_search_finds_a_vehicle_by_reference(ops_client):
response = ops_client.get("/api/v1/search", params={"q": "MO-001"})
assert response.status_code == 200
body = response.json()
match = next((r for r in body["results"] if r["type"] == "vehicle"), None)
assert match is not None
assert match["label"] == "MO-001"
assert match["link"] == "/vehicles/MO-001"
def test_search_finds_a_booking_by_reference(ops_client):
response = ops_client.get("/api/v1/search", params={"q": "BK-DEMO-RETURN"})
assert response.status_code == 200
match = next((r for r in response.json()["results"] if r["type"] == "booking"), None)
assert match is not None
assert match["link"] == "/bookings/BK-DEMO-RETURN"
def test_search_finds_a_data_quality_issue_for_operations_manager(ops_client):
response = ops_client.get("/api/v1/search", params={"q": "DQ-DEMO-OVERLAP"})
assert response.status_code == 200
match = next((r for r in response.json()["results"] if r["type"] == "data_quality_issue"), None)
assert match is not None
assert match["link"] == "/data-quality/DQ-DEMO-OVERLAP"
def test_search_never_returns_data_quality_issues_for_rental_employee(employee_client):
response = employee_client.get("/api/v1/search", params={"q": "DQ-DEMO-OVERLAP"})
assert response.status_code == 200
assert all(r["type"] != "data_quality_issue" for r in response.json()["results"])
def test_search_section_result_visible_to_operations_manager(ops_client):
result = ops_client.get("/api/v1/search", params={"q": "audit"}).json()
assert any(r["type"] == "section" and r["link"] == "/audit" for r in result["results"])
def test_search_section_result_hidden_from_rental_employee(employee_client):
result = employee_client.get("/api/v1/search", params={"q": "audit"}).json()
assert all(r["link"] != "/audit" for r in result["results"])
def test_search_never_returns_customer_results(ops_client):
response = ops_client.get("/api/v1/search", params={"q": "CUS-0012"})
assert response.status_code == 200
assert all(r["type"] != "customer" for r in response.json()["results"])
def test_search_no_match_returns_empty_results(ops_client):
response = ops_client.get("/api/v1/search", params={"q": "zzz-no-such-thing-zzz"})
assert response.status_code == 200
assert response.json()["results"] == []