| 1 | from __future__ import annotations |
| 2 | |
| 3 | import pytest |
| 4 | from mcp import Client |
| 5 | |
| 6 | from app.server import create_mcp_server |
| 7 | from app.settings import McpGatewaySettings |
| 8 | from conftest import INSTRUMENT_ID |
| 9 | |
| 10 | |
| 11 | def test_local_configuration_loads_without_azure(monkeypatch) -> None: |
| 12 | for name in ( |
| 13 | "AZURE_CLIENT_ID", |
| 14 | "AZURE_CLIENT_SECRET", |
| 15 | "AZURE_TENANT_ID", |
| 16 | "KEY_VAULT_NAME", |
| 17 | "OTEL_EXPORTER_OTLP_ENDPOINT", |
| 18 | ): |
| 19 | monkeypatch.delenv(name, raising=False) |
| 20 | settings = McpGatewaySettings(AIP_ENVIRONMENT="LOCAL") |
| 21 | assert settings.environment == "LOCAL" |
| 22 | assert settings.transport == "stdio" |
| 23 | assert settings.external_providers_enabled is False |
| 24 | assert settings.auth_context().authentication_type.value == "LOCAL_SERVICE" |
| 25 | |
| 26 | |
| 27 | def test_azure_configuration_uses_workload_identity_without_client_secret() -> None: |
| 28 | settings = McpGatewaySettings( |
| 29 | AIP_ENVIRONMENT="AZURE", |
| 30 | AIP_MCP_SERVICE_IDENTITY="mcp-gateway", |
| 31 | AIP_MCP_RESEARCH_BASE_URL="http://research-engine", |
| 32 | ) |
| 33 | serialized = str(settings.model_dump()).upper() |
| 34 | assert settings.auth_context().authentication_type.value == "WORKLOAD_IDENTITY" |
| 35 | assert "CLIENT_SECRET" not in serialized |
| 36 | assert settings.external_providers_enabled is False |
| 37 | |
| 38 | |
| 39 | @pytest.mark.asyncio |
| 40 | async def test_official_sdk_initializes_lists_schemas_and_calls_tool(container) -> None: |
| 41 | server = create_mcp_server(container) |
| 42 | async with Client(server) as client: |
| 43 | listing = await client.list_tools() |
| 44 | tools = {tool.name: tool for tool in listing.tools} |
| 45 | assert len(tools) == 9 |
| 46 | assert set(tools) == set(container.registry.tools) |
| 47 | assert "globalInstrumentId" in tools["get_research_readiness"].input_schema["properties"] |
| 48 | result = await client.call_tool( |
| 49 | "get_research_readiness", {"globalInstrumentId": str(INSTRUMENT_ID)} |
| 50 | ) |
| 51 | assert result.is_error is False |
| 52 | assert result.structured_content["ok"] is True |
| 53 | assert result.structured_content["data"]["overallStatus"] == "READY" |
| 54 | unknown = await client.call_tool("not_a_tool", {}) |
| 55 | assert unknown.is_error is True |
| 56 | assert unknown.structured_content["error"]["code"] == "MCP_TOOL_NOT_FOUND" |
| 57 | invalid = await client.call_tool("get_research_readiness", {}) |
| 58 | assert invalid.is_error is True |
| 59 | assert invalid.structured_content["error"]["code"] == "INVALID_ARGUMENT" |
| 60 | assert container.reader.closed is True |