"""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()