from __future__ import annotations import unittest from base64 import b64decode from uuid import UUID from services.control_api.contextforge.client import ContextForgeClient from services.control_api.contextforge.errors import ContextForgeError from services.control_api.contextforge.fake import FakeContextForge from services.control_api.contextforge.models import ( ContextForgeSettings, GatewayDesiredState, VirtualServerDesiredState, ) class ContextForgeAdapterTests(unittest.TestCase): def setUp(self) -> None: self.fake = FakeContextForge() settings = ContextForgeSettings( base_url="https://fake.contextforge.test", auth_token_env="CTX_TOKEN", # noqa: S106 retries=2, ) self.client = ContextForgeClient( settings, transport=self.fake.transport(), environ={"CTX_TOKEN": "fake-token-never-logged"}, ) self.desired = GatewayDesiredState( slug="knowledge-dev", url="https://knowledge.example.test/mcp", description="Knowledge test gateway", ) def tearDown(self) -> None: self.client.close() def test_health_capabilities_idempotent_crud_test_and_sanitized_inventory(self) -> None: self.assertEqual(self.client.health()["status"], "healthy") self.assertEqual(self.client.version(), "1.0.6") settings_from_env = ContextForgeSettings.from_env( { "HUB_CONTEXTFORGE_URL": "https://fake.contextforge.test", "HUB_CONTEXTFORGE_TOKEN_ENV": "CTX_TOKEN", } ) self.assertEqual(settings_from_env.auth_token_env, "CTX_TOKEN") capabilities = self.client.capabilities() self.assertTrue(capabilities.health) first = self.client.ensure_gateway(self.desired) second = self.client.ensure_gateway(self.desired) self.assertEqual(first.external_id, second.external_id) self.assertEqual(len(self.fake.gateways), 1) self.fake.tools[first.external_id] = [ { "id": "11111111-1111-4111-8111-111111111111", "name": "knowledge-dev-knowledge-document-search", "originalName": "knowledge.document.search", "customName": "knowledge.document.search", } ] gateway_tools = self.client.list_gateway_tools(first) self.assertEqual(gateway_tools[0].name, "knowledge.document.search") self.fake.prompts[first.external_id] = [ { "id": "44444444-4444-4444-8444-444444444444", "name": "project-context-pack", } ] gateway_prompts = self.client.list_gateway_prompts(first) self.assertEqual(gateway_prompts[0].name, "project-context-pack") desired_server = VirtualServerDesiredState( external_id=UUID("22222222-2222-4222-8222-222222222222"), name="ludarium-prod-readonly", description="Scoped production server", associated_tool_ids=(gateway_tools[0].external_id,), associated_prompt_ids=(gateway_prompts[0].external_id,), ) server = self.client.ensure_virtual_server(desired_server) self.assertEqual(server.associated_tool_ids, desired_server.associated_tool_ids) self.assertEqual(server.associated_prompt_ids, desired_server.associated_prompt_ids) self.assertEqual(server.visibility, "public") self.assertEqual( self.client.ensure_virtual_server(desired_server).external_id, server.external_id ) self.fake.servers[server.external_id]["visibility"] = "private" repaired = self.client.ensure_virtual_server(desired_server) self.assertEqual(repaired.visibility, "public") self.assertTrue(all(not path.startswith("/v1/") for _, path, _ in self.fake.requests)) self.assertNotIn("external_id", first.snapshot.model_dump()) self.assertNotIn(first.external_id, first.snapshot.model_dump_json()) listed = self.client.list_gateways() self.assertEqual(listed[0].slug, "knowledge-dev") result = self.client.test_gateway(first) self.assertTrue(result.reachable) by_slug = self.client.test_gateway_by_slug("knowledge-dev") self.assertTrue(by_slug.reachable) missing = self.client.test_gateway_by_slug("not-registered") self.assertFalse(missing.reachable) self.assertEqual(missing.status, "missing") updated = self.client.update_gateway( first, self.desired.model_copy(update={"description": "Updated safely"}), ) self.assertEqual(updated.snapshot.description, "Updated safely") inventory = self.client.inventory() self.assertEqual(inventory.gateways[0]["slug"], "knowledge-dev") self.assertIn("external_id_digest", inventory.gateways[0]) self.assertNotIn("external_id", inventory.gateways[0]) self.client.delete_gateway(updated) self.assertEqual(self.client.list_gateways(), ()) self.assertTrue( all("fake-token-never-logged" not in str(request) for request in self.fake.requests) ) def test_refresh_repopulates_stale_gateway_tools_and_prompts(self) -> None: gateway = self.client.ensure_gateway( self.desired.model_copy( update={"slug": "itworx-gitea", "url": "https://gitea.example.test/mcp"} ) ) self.fake.tools[gateway.external_id] = [] self.fake.prompts[gateway.external_id] = [] self.client.refresh_gateway(gateway) self.assertIn( "gitea.pullrequest.list", {tool.name for tool in self.client.list_gateway_tools(gateway)}, ) self.assertEqual( {prompt.name for prompt in self.client.list_gateway_prompts(gateway)}, { "repository-engineer", "repository-change-review", "repository-maintenance", "portfolio-engineer", "release-readiness", }, ) def test_ensure_gateway_reactivates_runtime_deactivated_gateway(self) -> None: first = self.client.ensure_gateway(self.desired) self.fake.gateways[first.external_id]["status"] = "offline" recovered = self.client.ensure_gateway(self.desired) self.assertEqual(recovered.external_id, first.external_id) self.assertEqual(recovered.snapshot.status, "active") self.assertTrue( any( method == "PUT" and path == f"/gateways/{first.external_id}" for method, path, _ in self.fake.requests ) ) def test_uses_original_tool_name_for_catalog_allowlist_matching(self) -> None: gateway = self.client.ensure_gateway(self.desired) self.fake.tools[gateway.external_id] = [ { "id": "33333333-3333-4333-8333-333333333333", "name": "knowledge-dev-knowledge-document-search", "originalName": "knowledge.document.search", "customName": "knowledge.document.search", } ] tools = self.client.list_gateway_tools(gateway) self.assertEqual(len(tools), 1) self.assertEqual(tools[0].name, "knowledge.document.search") self.assertEqual(tools[0].published_name, "knowledge.document.search") self.assertEqual(tools[0].external_id, "33333333-3333-4333-8333-333333333333") def test_canonicalizes_contextforge_prefixed_tool_name_once(self) -> None: gateway = self.client.ensure_gateway(self.desired) external_id = "33333333-3333-4333-8333-333333333333" self.fake.tools[gateway.external_id] = [ { "id": external_id, "name": "knowledge-dev-knowledge-document-search", "originalName": "knowledge.document.search", "customName": "knowledge-dev-knowledge-document-search", } ] canonical = self.client.ensure_canonical_tool_names(self.client.list_gateway_tools(gateway)) self.assertEqual(canonical[0].published_name, "knowledge.document.search") self.assertEqual( self.fake.tools[gateway.external_id][0]["customName"], "knowledge.document.search", ) self.client.ensure_canonical_tool_names(canonical) updates = [ request for request in self.fake.requests if request[0] == "PUT" and request[1] == f"/tools/{external_id}" ] self.assertEqual(len(updates), 1) def test_reads_gateway_tool_input_schema_in_both_supported_key_styles(self) -> None: gateway = self.client.ensure_gateway(self.desired) self.fake.tools[gateway.external_id] = [ { "id": "33333333-3333-4333-8333-333333333333", "name": "knowledge.document.search", "input_schema": {"type": "object", "required": ["query"]}, } ] tool = self.client.list_gateway_tools(gateway)[0] self.assertEqual(tool.input_schema, {"type": "object", "required": ["query"]}) def test_basic_compatibility_auth_is_encoded_and_never_logged(self) -> None: settings = ContextForgeSettings( base_url="https://fake.contextforge.test", auth_token_env="CTX_PASSWORD", # noqa: S106 auth_scheme="Basic", basic_username="bootstrap-admin", ) client = ContextForgeClient( settings, transport=self.fake.transport(), environ={"CTX_PASSWORD": "password-never-logged"}, ) try: headers = client._auth_headers() encoded = headers["Authorization"].removeprefix("Basic ") self.assertEqual( b64decode(encoded).decode("utf-8"), "bootstrap-admin:password-never-logged", ) client.health() self.assertNotIn("password-never-logged", str(self.fake.requests)) finally: client.close() def test_missing_auth_and_stable_gateway_error_mapping(self) -> None: settings = ContextForgeSettings( base_url="https://fake.contextforge.test", auth_token_env="MISSING_TOKEN", # noqa: S106 ) missing = ContextForgeClient(settings, transport=self.fake.transport(), environ={}) try: with self.assertRaises(ContextForgeError) as raised: missing.health() self.assertEqual(raised.exception.code, "gateway_credentials_missing") finally: missing.close() no_retry_settings = settings.model_copy(update={"retries": 0}) no_retry = ContextForgeClient( no_retry_settings, transport=self.fake.transport(), environ={"MISSING_TOKEN": "token"}, ) try: self.fake.fail_next = 503 with self.assertRaises(ContextForgeError) as raised: no_retry.health() self.assertEqual(raised.exception.code, "gateway_unavailable") self.assertTrue(raised.exception.retryable) finally: no_retry.close() if __name__ == "__main__": unittest.main()