352 lines
15 KiB
Python
352 lines
15 KiB
Python
"""Contract and adversarial tests for the MobilityOps (Fleet Ops) connector."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
from uuid import UUID, uuid4
|
|
|
|
from connectors.mobilityops import FakeMobilityOpsServer, MobilityOpsConnector
|
|
from packages.connector_kit import (
|
|
ConnectorContext,
|
|
MobilityOpsClient,
|
|
MobilityOpsError,
|
|
MobilityOpsSettings,
|
|
MobilityOpsStaticTokenProvider,
|
|
)
|
|
|
|
PROJECT_ID = UUID("11111111-1111-1111-1111-111111111111")
|
|
TRACE_ID = "hub-trace-0123456789"
|
|
|
|
|
|
def context(
|
|
project_id: UUID = PROJECT_ID,
|
|
slug: str = "mobilityops",
|
|
correlation_id: str | None = TRACE_ID,
|
|
) -> ConnectorContext:
|
|
return ConnectorContext(
|
|
workspace_id=uuid4(),
|
|
project_id=project_id,
|
|
project_slug=slug,
|
|
environment="test",
|
|
correlation_id=correlation_id,
|
|
)
|
|
|
|
|
|
def connector(
|
|
server: FakeMobilityOpsServer | None = None,
|
|
) -> tuple[MobilityOpsConnector, FakeMobilityOpsServer]:
|
|
fixture = server or FakeMobilityOpsServer()
|
|
settings = MobilityOpsSettings(
|
|
base_url="http://mobilityops.test",
|
|
token_reference="docker-secret://mobilityops_service_token", # noqa: S106
|
|
allow_insecure_http_for_tests=True,
|
|
)
|
|
client = MobilityOpsClient(
|
|
settings,
|
|
MobilityOpsStaticTokenProvider("fixture-token-must-not-leak"),
|
|
transport=fixture.transport(),
|
|
)
|
|
return MobilityOpsConnector(client), fixture
|
|
|
|
|
|
class MobilityOpsConnectorTests(unittest.TestCase):
|
|
def test_operations_summary_maps_metrics(self) -> None:
|
|
connector_instance, _ = connector()
|
|
try:
|
|
result = connector_instance.operations_summary(context())
|
|
self.assertEqual(result["data"]["tenant"], "Northstar Mobility Demo")
|
|
self.assertEqual(result["data"]["metrics"]["available"], 21)
|
|
self.assertEqual(result["data"]["metrics"]["openQualityIssues"], 3)
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
def test_attention_list_maps_items_and_respects_limit(self) -> None:
|
|
connector_instance, fixture = connector()
|
|
try:
|
|
result = connector_instance.attention_list(context(), limit=1)
|
|
self.assertEqual(len(result["data"]["items"]), 1)
|
|
self.assertEqual(result["data"]["items"][0]["vehicleRef"], "MO-016")
|
|
self.assertEqual(result["data"]["items"][0]["severity"], "high")
|
|
self.assertTrue(all(method == "GET" for method in fixture.recorded_methods))
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
def test_attention_list_rejects_invalid_severity_and_limit(self) -> None:
|
|
connector_instance, _ = connector()
|
|
try:
|
|
with self.assertRaises(ValueError):
|
|
connector_instance.attention_list(context(), minimum_severity="urgent")
|
|
with self.assertRaises(ValueError):
|
|
connector_instance.attention_list(context(), limit=0)
|
|
with self.assertRaises(ValueError):
|
|
connector_instance.attention_list(context(), limit=51)
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
def test_vehicle_get_maps_fields(self) -> None:
|
|
connector_instance, _ = connector()
|
|
try:
|
|
result = connector_instance.vehicle_get(context(), "MO-016")
|
|
self.assertEqual(result["data"]["publicRef"], "MO-016")
|
|
self.assertEqual(result["data"]["make"], "Volkswagen")
|
|
self.assertIsNone(result["data"]["currentBookingRef"])
|
|
self.assertEqual(result["meta"]["source"], "mobilityops:MO-016")
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
def test_vehicle_get_rejects_path_unsafe_reference(self) -> None:
|
|
connector_instance, _ = connector()
|
|
try:
|
|
for vehicle_ref in ("../secret", "MO/016", "MO 016", ""):
|
|
with self.subTest(vehicle_ref=vehicle_ref), self.assertRaises(MobilityOpsError):
|
|
connector_instance.vehicle_get(context(), vehicle_ref)
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
def test_vehicle_not_found_is_a_stable_error(self) -> None:
|
|
connector_instance, _ = connector(FakeMobilityOpsServer(vehicle_not_found=True))
|
|
try:
|
|
with self.assertRaisesRegex(MobilityOpsError, "not found"):
|
|
connector_instance.vehicle_get(context(), "MO-999")
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
def test_knowledge_search_maps_grounded_answer_and_sources(self) -> None:
|
|
connector_instance, _ = connector()
|
|
try:
|
|
result = connector_instance.knowledge_search(
|
|
context(), "What must I do about vehicle damage?"
|
|
)
|
|
self.assertEqual(result["data"]["evidenceState"], "grounded")
|
|
self.assertTrue(result["data"]["answer"])
|
|
self.assertEqual(len(result["data"]["sources"]), 1)
|
|
self.assertEqual(result["data"]["sources"][0]["documentId"], "damage-procedure")
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
def test_knowledge_search_rejects_out_of_budget_input(self) -> None:
|
|
connector_instance, _ = connector()
|
|
try:
|
|
with self.assertRaises(ValueError):
|
|
connector_instance.knowledge_search(context(), "hi")
|
|
with self.assertRaises(ValueError):
|
|
connector_instance.knowledge_search(context(), "a valid question", max_sources=9)
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
def test_token_and_response_headers_are_not_exposed(self) -> None:
|
|
connector_instance, fixture = connector()
|
|
try:
|
|
output = repr(connector_instance.vehicle_get(context(), "MO-016"))
|
|
self.assertNotIn("fixture-token-must-not-leak", output)
|
|
self.assertEqual(fixture.recorded_service_tokens[0], "fixture-token-must-not-leak")
|
|
self.assertNotIn("x-service-token", output.lower())
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
def test_client_id_is_derived_from_context_for_audit_traceability(self) -> None:
|
|
connector_instance, fixture = connector()
|
|
try:
|
|
connector_instance.operations_summary(context(slug="mobilityops"))
|
|
self.assertIn("mobilityops", fixture.recorded_client_ids[0])
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
def test_unauthorized_service_token_is_a_stable_error(self) -> None:
|
|
connector_instance, _ = connector(FakeMobilityOpsServer(unauthorized=True))
|
|
try:
|
|
with self.assertRaisesRegex(MobilityOpsError, "denied"):
|
|
connector_instance.operations_summary(context())
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
def test_redirect_to_other_origin_is_denied(self) -> None:
|
|
connector_instance, fixture = connector(
|
|
FakeMobilityOpsServer(redirect_to="http://169.254.169.254/latest/meta-data")
|
|
)
|
|
try:
|
|
with self.assertRaisesRegex(MobilityOpsError, "leaves the configured origin"):
|
|
connector_instance.operations_summary(context())
|
|
self.assertEqual(len(fixture.recorded_methods), 1)
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
def test_publication_is_read_only_and_cataloged(self) -> None:
|
|
connector_instance, _ = connector()
|
|
try:
|
|
names = {tool.canonical_name for tool in connector_instance.tools()}
|
|
self.assertEqual(
|
|
names,
|
|
{
|
|
"mobilityops.operations.summary",
|
|
"mobilityops.attention.list",
|
|
"mobilityops.vehicle.get",
|
|
"mobilityops.knowledge.search",
|
|
},
|
|
)
|
|
self.assertTrue(all(tool.read_only for tool in connector_instance.tools()))
|
|
self.assertTrue(
|
|
all("write" not in tool.canonical_name for tool in connector_instance.tools())
|
|
)
|
|
self.assertEqual(connector_instance.resources(), ())
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
def test_secret_reference_and_transport_policy_are_strict(self) -> None:
|
|
with self.assertRaises(ValueError):
|
|
MobilityOpsSettings(
|
|
base_url="http://mobilityops.test",
|
|
token_reference="raw-token", # noqa: S106
|
|
)
|
|
with self.assertRaises(ValueError):
|
|
MobilityOpsSettings(
|
|
base_url="https://user:pass@mobilityops.test",
|
|
token_reference="docker-secret://mobilityops", # noqa: S106
|
|
)
|
|
|
|
def test_knowledge_search_sends_the_requested_locale(self) -> None:
|
|
for locale in ("nl-BE", "en-GB", "fr-BE"):
|
|
with self.subTest(locale=locale):
|
|
connector_instance, fixture = connector()
|
|
try:
|
|
connector_instance.knowledge_search(
|
|
context(), "What must I do about vehicle damage?", locale=locale
|
|
)
|
|
self.assertEqual(fixture.recorded_locales, [locale])
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
def test_knowledge_search_defaults_to_en_gb_for_older_clients(self) -> None:
|
|
connector_instance, fixture = connector()
|
|
try:
|
|
connector_instance.knowledge_search(context(), "What about vehicle damage?")
|
|
self.assertEqual(fixture.recorded_locales, ["en-GB"])
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
def test_knowledge_search_rejects_an_unsupported_locale(self) -> None:
|
|
connector_instance, fixture = connector()
|
|
try:
|
|
with self.assertRaises(ValueError):
|
|
connector_instance.knowledge_search(
|
|
context(), "What about vehicle damage?", locale="de-DE"
|
|
)
|
|
self.assertEqual(fixture.recorded_methods, [])
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
def test_inbound_hub_trace_id_is_forwarded_on_every_call(self) -> None:
|
|
connector_instance, fixture = connector()
|
|
try:
|
|
connector_instance.operations_summary(context())
|
|
connector_instance.attention_list(context())
|
|
connector_instance.vehicle_get(context(), "MO-016")
|
|
connector_instance.knowledge_search(context(), "What about vehicle damage?")
|
|
self.assertEqual(fixture.recorded_correlation_ids, [TRACE_ID] * 4)
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
def test_no_correlation_header_is_invented_when_the_hub_has_none(self) -> None:
|
|
connector_instance, fixture = connector()
|
|
try:
|
|
connector_instance.operations_summary(context(correlation_id=None))
|
|
self.assertEqual(fixture.recorded_correlation_ids, [""])
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
def test_knowledge_search_preserves_provider_and_correlation_metadata(self) -> None:
|
|
connector_instance, _ = connector()
|
|
try:
|
|
result = connector_instance.knowledge_search(
|
|
context(), "What must I do about vehicle damage?"
|
|
)
|
|
self.assertEqual(result["data"]["provider"], "demo")
|
|
self.assertEqual(result["data"]["correlationId"], TRACE_ID)
|
|
self.assertEqual(result["data"]["evidenceState"], "grounded")
|
|
self.assertEqual(len(result["data"]["sources"]), 1)
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
def test_absent_provider_metadata_is_reported_as_null_not_invented(self) -> None:
|
|
fixture = FakeMobilityOpsServer()
|
|
fixture._answer = { # noqa: SLF001 - fixture-owned payload, intentional.
|
|
"answer": "No provider field on this response.",
|
|
"evidence_state": "insufficient",
|
|
"sources": [],
|
|
"correlation_id": "",
|
|
}
|
|
connector_instance, _ = connector(fixture)
|
|
try:
|
|
result = connector_instance.knowledge_search(
|
|
context(correlation_id=None), "What about vehicle damage?"
|
|
)
|
|
self.assertIsNone(result["data"]["provider"])
|
|
self.assertIsNone(result["data"]["correlationId"])
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
def test_readiness_is_ready_only_when_the_upstream_answers(self) -> None:
|
|
connector_instance, fixture = connector()
|
|
try:
|
|
snapshot = connector_instance.readiness()
|
|
self.assertEqual(snapshot["status"], "ready")
|
|
self.assertEqual(fixture.recorded_methods, ["GET"])
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
def test_readiness_is_unavailable_when_the_upstream_cannot_be_reached(self) -> None:
|
|
connector_instance, _ = connector(FakeMobilityOpsServer(unreachable=True))
|
|
try:
|
|
snapshot = connector_instance.readiness()
|
|
self.assertEqual(snapshot["status"], "unavailable")
|
|
self.assertEqual(snapshot["reason"], "MOBILITYOPS_UNAVAILABLE")
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
def test_readiness_is_degraded_when_the_upstream_denies_the_service_token(self) -> None:
|
|
connector_instance, _ = connector(FakeMobilityOpsServer(unauthorized=True))
|
|
try:
|
|
snapshot = connector_instance.readiness()
|
|
self.assertEqual(snapshot["status"], "degraded")
|
|
self.assertEqual(snapshot["reason"], "MOBILITYOPS_ACCESS_DENIED")
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
def test_readiness_never_exposes_a_secret_or_raw_exception(self) -> None:
|
|
for fixture in (
|
|
FakeMobilityOpsServer(),
|
|
FakeMobilityOpsServer(unauthorized=True),
|
|
FakeMobilityOpsServer(unreachable=True),
|
|
):
|
|
with self.subTest(fixture=fixture):
|
|
connector_instance, _ = connector(fixture)
|
|
try:
|
|
output = repr(connector_instance.readiness())
|
|
self.assertNotIn("fixture-token-must-not-leak", output)
|
|
self.assertNotIn("Traceback", output)
|
|
self.assertNotIn("httpx", output.lower())
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
def test_health_stays_process_liveness_and_is_not_an_upstream_probe(self) -> None:
|
|
connector_instance, fixture = connector(FakeMobilityOpsServer(unreachable=True))
|
|
try:
|
|
self.assertEqual(connector_instance.health()["status"], "healthy")
|
|
self.assertEqual(fixture.recorded_methods, [])
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
def test_invoke_dispatches_by_canonical_name_and_rejects_unknown_tool(self) -> None:
|
|
connector_instance, _ = connector()
|
|
try:
|
|
result = connector_instance.invoke("mobilityops.operations.summary", context())
|
|
self.assertEqual(result["data"]["tenant"], "Northstar Mobility Demo")
|
|
with self.assertRaises(ValueError):
|
|
connector_instance.invoke("mobilityops.vehicle.delete", context())
|
|
finally:
|
|
connector_instance.client.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|