130 lines
5.1 KiB
Python
130 lines
5.1 KiB
Python
"""Deterministic in-memory MobilityOps HTTP transport for contract tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from urllib.parse import parse_qs, urlparse
|
|
|
|
import httpx
|
|
|
|
|
|
class FakeMobilityOpsServer:
|
|
"""A tiny MobilityOps API fixture with observable requests and no network."""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
unauthorized: bool = False,
|
|
redirect_to: str | None = None,
|
|
vehicle_not_found: bool = False,
|
|
unreachable: bool = False,
|
|
) -> None:
|
|
self.unauthorized = unauthorized
|
|
self.redirect_to = redirect_to
|
|
self.vehicle_not_found = vehicle_not_found
|
|
self.unreachable = unreachable
|
|
self.recorded_methods: list[str] = []
|
|
self.recorded_service_tokens: list[str] = []
|
|
self.recorded_client_ids: list[str] = []
|
|
self.recorded_correlation_ids: list[str] = []
|
|
self.recorded_locales: list[str | None] = []
|
|
self._summary = {
|
|
"tenant": "Northstar Mobility Demo",
|
|
"metrics": {
|
|
"available": 21,
|
|
"rented": 11,
|
|
"cleaning": 6,
|
|
"maintenance": 5,
|
|
"blocked": 7,
|
|
"open_quality_issues": 3,
|
|
"pending_or_failed_workflows": 1,
|
|
},
|
|
}
|
|
self._attention = [
|
|
{
|
|
"vehicle_ref": "MO-016",
|
|
"severity": "high",
|
|
"rule_type": "vehicle_status_conflict",
|
|
"summary": "Booking overlap detected for this vehicle.",
|
|
"detected_at": "2026-08-01T09:00:00Z",
|
|
"token": "fixture-token-must-not-leak",
|
|
}
|
|
]
|
|
self._vehicle = {
|
|
"public_ref": "MO-016",
|
|
"make": "Volkswagen",
|
|
"model": "ID.4",
|
|
"model_year": 2024,
|
|
"location": "Antwerp Depot",
|
|
"operational_status": "available",
|
|
"odometer_km": 18420,
|
|
"next_service_km": 40000,
|
|
"open_quality_issue_count": 1,
|
|
"current_booking_ref": None,
|
|
}
|
|
self._answer = {
|
|
"answer": "Report the damage and route the vehicle to maintenance.",
|
|
"evidence_state": "grounded",
|
|
"sources": [
|
|
{
|
|
"document_id": "damage-procedure",
|
|
"title": "Damage handling procedure",
|
|
"version": "1",
|
|
"section": "Detection",
|
|
"excerpt": "Inspect the vehicle for visible damage.",
|
|
}
|
|
],
|
|
"provider": "demo",
|
|
"correlation_id": "fixture-correlation-id",
|
|
}
|
|
|
|
def transport(self) -> httpx.MockTransport:
|
|
return httpx.MockTransport(self._handle)
|
|
|
|
def _handle(self, request: httpx.Request) -> httpx.Response:
|
|
self.recorded_methods.append(request.method)
|
|
self.recorded_service_tokens.append(request.headers.get("x-service-token", ""))
|
|
self.recorded_client_ids.append(request.headers.get("x-client-id", ""))
|
|
self.recorded_correlation_ids.append(request.headers.get("x-correlation-id", ""))
|
|
if self.unreachable:
|
|
raise httpx.ConnectError("fixture target is unreachable", request=request)
|
|
if self.redirect_to is not None:
|
|
return httpx.Response(302, headers={"location": self.redirect_to}, request=request)
|
|
if self.unauthorized:
|
|
return self._json(401, {"code": "UNAUTHORIZED_SERVICE"}, request)
|
|
|
|
parsed = urlparse(str(request.url))
|
|
path = parsed.path.removeprefix("/api/v1/integrations/mcp")
|
|
query = parse_qs(parsed.query)
|
|
|
|
if path == "/operations-summary" and request.method == "GET":
|
|
return self._json(200, self._summary, request)
|
|
if path == "/attention-vehicles" and request.method == "GET":
|
|
limit = int(query.get("limit", ["20"])[0])
|
|
return self._json(200, self._attention[:limit], request)
|
|
if path.startswith("/vehicles/") and request.method == "GET":
|
|
if self.vehicle_not_found:
|
|
return self._json(404, {"code": "VEHICLE_NOT_FOUND"}, request)
|
|
return self._json(200, self._vehicle, request, leaked_header=True)
|
|
if path == "/search-knowledge" and request.method == "POST":
|
|
body = json.loads(request.content or b"{}")
|
|
self.recorded_locales.append(body.get("locale"))
|
|
# MobilityOps echoes the correlation id it actually received.
|
|
answer = dict(self._answer)
|
|
answer["correlation_id"] = (
|
|
request.headers.get("x-correlation-id") or answer["correlation_id"]
|
|
)
|
|
return self._json(200, answer, request)
|
|
return self._json(404, {"code": "NOT_FOUND"}, request)
|
|
|
|
@staticmethod
|
|
def _json(
|
|
status: int, payload: object, request: httpx.Request, *, leaked_header: bool = False
|
|
) -> httpx.Response:
|
|
headers = {"content-type": "application/json"}
|
|
if leaked_header:
|
|
headers["x-fixture-token"] = "fixture-token-must-not-leak"
|
|
return httpx.Response(
|
|
status, headers=headers, content=json.dumps(payload).encode(), request=request
|
|
)
|