diff --git a/CHANGELOG.md b/CHANGELOG.md index ea197f037..8291058fa 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,6 +24,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- **Breaking:** MCP registry client now speaks the official MCP Registry v0.1 spec. **Self-hosted registries must serve `/v0.1/` paths -- registries serving only the legacy `/v0/` paths will return 404.** The public registry at `api.mcp.github.com` and the official `registry.modelcontextprotocol.io` are unaffected. Python API: `SimpleRegistryClient.get_server_info(server_id)` is renamed to `get_server(server_name, version="latest")` (the parameter is now a server *name* per the spec, not a UUID); the old name remains for one minor as a `DeprecationWarning` shim. The legacy UUID strategy in `find_server_by_reference` is removed -- the spec keys per-server lookup on serverName. v0.1 package fields (`identifier`, `registryType`, `runtimeHint`, `packageArguments`, `runtimeArguments`, `environmentVariables`) are normalized at the registry boundary to the snake_case shape adapters consume. Thanks @fassmus for the report. (#1337, closes #1210) - `--marketplace-output PATH` is now hidden from `--help` and emits a stderr deprecation warning; it auto-translates to `--marketplace-path claude=PATH`. Removal tracked in #1318. (#1317) - `extends: org` now correctly layers `dependencies.require` and `dependencies.deny` from the parent policy when the child omits the `dependencies:` block entirely; `None` signals "no opinion" (transparent) while `[]` signals explicit override. (#1290) - CI self-check job now uses `setup-only: true` + `apm audit --ci --no-drift` so managed files are not overwritten by `apm install` before `content-integrity` runs; documented the audit-only CI pattern and the install-before-audit blind spot in the enterprise and CI/CD guides. (#1291) diff --git a/docs/src/content/docs/reference/cli/mcp.md b/docs/src/content/docs/reference/cli/mcp.md index 477684829..1227a7c0f 100644 --- a/docs/src/content/docs/reference/cli/mcp.md +++ b/docs/src/content/docs/reference/cli/mcp.md @@ -145,6 +145,8 @@ export MCP_REGISTRY_URL=https://mcp.internal.example.com apm mcp list ``` +The registry must implement the [MCP Registry v0.1 spec](https://github.com/modelcontextprotocol/registry) (apm calls `/v0.1/servers/...`). Registries serving only the legacy `/v0/` paths will return 404. + ## Related - [`apm install`](../install/) -- canonical MCP install path. diff --git a/packages/apm-guide/.apm/skills/apm-usage/commands.md b/packages/apm-guide/.apm/skills/apm-usage/commands.md index 32b3e3af0..a3dc22b31 100644 --- a/packages/apm-guide/.apm/skills/apm-usage/commands.md +++ b/packages/apm-guide/.apm/skills/apm-usage/commands.md @@ -118,7 +118,7 @@ To build the marketplace, run `apm pack` (it reads `apm.yml` and writes `.claude | `apm mcp search QUERY` | Search MCP registry | `--limit N` | | `apm mcp show SERVER` | Show server details | -- | -Set `MCP_REGISTRY_URL` (default `https://api.mcp.github.com`) to point all `apm mcp` commands and `apm install --mcp` at a custom MCP registry. The URL is validated at startup and must use `https://`; set `MCP_REGISTRY_ALLOW_HTTP=1` to opt in to plaintext `http://` for development. When the override is set and the registry is unreachable during install pre-flight, APM fails closed. +Set `MCP_REGISTRY_URL` (default `https://api.mcp.github.com`) to point all `apm mcp` commands and `apm install --mcp` at a custom MCP registry. The URL is validated at startup and must use `https://`; set `MCP_REGISTRY_ALLOW_HTTP=1` to opt in to plaintext `http://` for development. The registry must implement the [MCP Registry v0.1 spec](https://github.com/modelcontextprotocol/registry) (apm calls `/v0.1/servers/...`); legacy `/v0/`-only registries will return 404. When the override is set and the registry is unreachable during install pre-flight, APM fails closed. ## Runtime management (experimental) diff --git a/src/apm_cli/registry/client.py b/src/apm_cli/registry/client.py index 37b15438c..bc346729e 100644 --- a/src/apm_cli/registry/client.py +++ b/src/apm_cli/registry/client.py @@ -2,8 +2,10 @@ import logging import os +import re +import warnings from typing import Any, Dict, List, Optional, Tuple # noqa: F401, UP035 -from urllib.parse import urlparse +from urllib.parse import quote, urlparse import requests @@ -20,6 +22,40 @@ def _safe_headers(response) -> dict[str, str]: _DEFAULT_REGISTRY_URL = "https://api.mcp.github.com" +# MCP Registry API version path prefix. Bumping this here is the +# single grep target the day v0.2 ships. See +# https://github.com/modelcontextprotocol/registry for the spec. +_V0_1_PREFIX = "/v0.1" + +# Allowlist for server names used as URL path segments. The MCP spec +# constrains names to reverse-DNS-style identifiers, optionally with a +# single ``/`` slug suffix. ``quote(name, safe='')`` already +# neutralises traversal/SSRF; this allowlist makes the threat model +# explicit at the call site so a future caller cannot bypass search +# and feed attacker-controlled strings into the path. +_SERVER_NAME_RE = re.compile(r"^[A-Za-z0-9._~-]+(/[A-Za-z0-9._~-]+)?$") + + +class ServerNotFoundError(ValueError): + """Raised when a server lookup against the registry returns 404. + + Carries the registry URL so the CLI boundary can render an + actionable hint about MCP Registry v0.1 spec compliance for + self-hosted registries. Inherits from ``ValueError`` so existing + ``except ValueError`` callers (e.g. ``find_server_by_reference``) + keep treating 404s as "not found" without code changes. + """ + + def __init__(self, server_name: str, registry_url: str) -> None: + self.server_name = server_name + self.registry_url = registry_url + super().__init__( + f"Server '{server_name}' not found in registry {registry_url}. " + f"If this is a self-hosted registry, verify it implements the " + f"MCP Registry v0.1 API (apm uses /v0.1/servers/...)." + ) + + # Network timeouts for registry HTTP calls. ``connect`` bounds the TCP # handshake (typo in --registry / unreachable host) so ``apm install`` # never hangs in CI; ``read`` bounds slow registries / proxies. @@ -96,6 +132,17 @@ def __init__(self, registry_url: str | None = None): f"Check MCP_REGISTRY_URL if set." ) + # Strip any embedded userinfo (``user:pass@``) before storing the URL so + # ``ServerNotFoundError`` and other diagnostics cannot leak credentials + # into terminal output or CI logs. Enterprise users sometimes set + # ``MCP_REGISTRY_URL=https://token:x-oauth@registry.corp/`` -- we still + # accept the URL (the credentials are passed via Authorization headers + # elsewhere), but we never echo them back. + if parsed.username or parsed.password: + host = parsed.hostname or "" + sanitized_netloc = host + (f":{parsed.port}" if parsed.port else "") + resolved = parsed._replace(netloc=sanitized_netloc).geturl().rstrip("/") + self.registry_url = resolved # True when the URL came from an explicit caller arg or MCP_REGISTRY_URL env var. # Consumed by validate_servers_exist() to fail-closed on overrides. @@ -217,17 +264,20 @@ def list_servers( ) -> tuple[list[dict[str, Any]], str | None]: """List all available servers in the registry. + Calls ``GET /v0.1/servers`` per the MCP Registry spec. + Args: limit (int, optional): Maximum number of entries to return. Defaults to 100. cursor (str, optional): Pagination cursor for retrieving next set of results. Returns: - Tuple[List[Dict[str, Any]], Optional[str]]: List of server metadata dictionaries and the next cursor if available. + Tuple[List[Dict[str, Any]], Optional[str]]: List of server metadata + dictionaries and the next cursor if available. Raises: requests.RequestException: If the request fails. """ - url = f"{self.registry_url}/v0/servers" + url = f"{self.registry_url}{_V0_1_PREFIX}/servers" params = {} if limit is not None: @@ -238,25 +288,27 @@ def list_servers( data, _hdrs = self._cached_get_json(url, params=params) data = data or {} - # Extract servers - they're nested under "server" key in each item - raw_servers = data.get("servers", []) - servers = [] - for item in raw_servers: - if "server" in item: - servers.append(item["server"]) - else: - servers.append(item) # Fallback for different structure + servers = self._unwrap_server_list(data) metadata = data.get("metadata", {}) - next_cursor = metadata.get("next_cursor") + # Spec is camelCase ``nextCursor``; ``next_cursor`` accepted as a + # transitional kindness for in-tree mock fixtures only. + # TODO(v0.1): drop legacy snake_case once fixtures migrate. + next_cursor = metadata.get("nextCursor") or metadata.get("next_cursor") return servers, next_cursor def search_servers(self, query: str) -> list[dict[str, Any]]: - """Search for servers in the registry using the API search endpoint. + """Search for servers in the registry using the spec ``?search=`` query param. + + Calls ``GET /v0.1/servers?search=`` per the MCP Registry spec + (case-insensitive substring match on server names). Args: - query (str): Search query string. + query (str): Search query string. The full reference is passed + through to the registry; spec-compliant registries do + substring matching on names so ``io.github.foo/bar`` and + ``bar`` both match ``io.github.foo/bar``. Returns: List[Dict[str, Any]]: List of matching server metadata dictionaries. @@ -264,62 +316,160 @@ def search_servers(self, query: str) -> list[dict[str, Any]]: Raises: requests.RequestException: If the request fails. """ - # The MCP Registry API now only accepts repository names (e.g., "github-mcp-server") - # If the query looks like a full identifier (e.g., "io.github.github/github-mcp-server"), - # extract the repository name for the search - search_query = self._extract_repository_name(query) - - url = f"{self.registry_url}/v0/servers/search" - params = {"q": search_query} + url = f"{self.registry_url}{_V0_1_PREFIX}/servers" + params = {"search": query} data, _hdrs = self._cached_get_json(url, params=params) data = data or {} - # Extract servers - they're nested under "server" key in each item + return self._unwrap_server_list(data) + + @staticmethod + def _unwrap_server_list(data: dict[str, Any]) -> list[dict[str, Any]]: + """Strict v0.1 unwrap of the ``servers`` array. + + Each entry is expected to carry a nested ``server`` object per + the spec. We deliberately do NOT fall back to flat shapes -- a + non-conformant registry should fail loudly here, not silently + produce half-shaped dicts that explode three call frames away + in conflict detection. + """ raw_servers = data.get("servers", []) servers = [] for item in raw_servers: - if "server" in item: - servers.append(item["server"]) - else: - servers.append(item) # Fallback for different structure - + if not isinstance(item, dict) or "server" not in item: + raise ValueError( + "Registry returned a non-spec list entry (missing 'server' key); " + "expected MCP Registry v0.1 response shape." + ) + servers.append(item["server"]) return servers - def get_server_info(self, server_id: str) -> dict[str, Any]: - """Get detailed information about a specific server. + # Map of v0.1 spec package field names (camelCase / renamed) to the + # legacy snake_case shape that adapters in src/apm_cli/adapters/client/ + # consume (e.g. package.get("name"), package.get("runtime_hint")). + # The registry boundary normalizes inbound packages so adapters keep + # working without per-adapter rewrites. See #1210 review feedback: + # without this, registry-resolved installs produced "npx -y None". + _V0_1_PACKAGE_FIELD_ALIASES: tuple[tuple[str, str], ...] = ( + ("identifier", "name"), + ("registryType", "registry_name"), + ("registryBaseUrl", "registry_base_url"), + ("runtimeHint", "runtime_hint"), + ("packageArguments", "package_arguments"), + ("runtimeArguments", "runtime_arguments"), + ("environmentVariables", "environment_variables"), + ) + + @classmethod + def _normalize_v0_1_package(cls, package: dict[str, Any]) -> dict[str, Any]: + """Backfill legacy snake_case keys from v0.1 camelCase aliases. + + Only writes the legacy key when the package does not already + carry one, so registries that emit both shapes (or the legacy + shape only) are unaffected. This is a one-way bridge: the + camelCase key is preserved so callers that have already migrated + keep working. + """ + if not isinstance(package, dict): + return package + normalized = dict(package) + for v01_key, legacy_key in cls._V0_1_PACKAGE_FIELD_ALIASES: + if legacy_key not in normalized and v01_key in normalized: + normalized[legacy_key] = normalized[v01_key] + return normalized + + @classmethod + def _normalize_v0_1_server(cls, server: dict[str, Any]) -> dict[str, Any]: + """Apply package-shape normalization to a server detail dict. + + Returns a shallow copy with each entry of ``packages`` normalized + via :meth:`_normalize_v0_1_package`. The input dict is not mutated, + matching the copy semantics of the sibling normalizer. + """ + if not isinstance(server, dict): + return server + normalized = dict(server) + packages = normalized.get("packages") + if isinstance(packages, list) and packages: + normalized["packages"] = [cls._normalize_v0_1_package(p) for p in packages] + return normalized + + def get_server(self, server_name: str, version: str = "latest") -> dict[str, Any]: + """Get detailed information about a specific server version. + + Calls ``GET /v0.1/servers/{urlencoded-serverName}/versions/{version}`` + per the MCP Registry spec. The default ``version="latest"`` covers + 99% of callers; pin to a specific version string for reproducibility. Args: - server_id (str): ID of the server. + server_name (str): Full server name (e.g. ``io.github.foo/bar``). + Validated against the spec name shape and URL-encoded. + version (str, optional): Version string or ``"latest"``. Defaults + to ``"latest"``. Returns: - Dict[str, Any]: Server metadata dictionary. + Dict[str, Any]: Server metadata dictionary (the unwrapped + contents of the response's ``server`` field, with any + top-level siblings merged in). Raises: - requests.RequestException: If the request fails. - ValueError: If the server is not found. + ValueError: If the server name does not match the spec shape. + ServerNotFoundError: If the registry returns 404. + requests.RequestException: If the request fails for other reasons. """ - url = f"{self.registry_url}/v0/servers/{server_id}" - data, _hdrs = self._cached_get_json(url) + if not _SERVER_NAME_RE.match(server_name or ""): + raise ValueError( + f"Invalid server name {server_name!r}: expected MCP spec shape " + f"(reverse-DNS identifier, optionally with a single '/' suffix)." + ) + + encoded_name = quote(server_name, safe="") + encoded_version = quote(version, safe="") + url = f"{self.registry_url}{_V0_1_PREFIX}/servers/{encoded_name}/versions/{encoded_version}" + + try: + data, _hdrs = self._cached_get_json(url) + except requests.HTTPError as exc: + response = getattr(exc, "response", None) + if response is not None and response.status_code == 404: + raise ServerNotFoundError(server_name, self.registry_url) from exc + raise + data = data or {} - # Return the complete response including x-github and other metadata - # but ensure the main server info is accessible at the top level + # Return the complete response including _meta and other top-level + # metadata, but ensure the main server info is accessible at the top level. if "server" in data: - # Merge server info to top level while preserving x-github and other sections result = data["server"].copy() for key, value in data.items(): if key != "server": result[key] = value - if not result: - raise ValueError(f"Server '{server_id}' not found in registry") + raise ServerNotFoundError(server_name, self.registry_url) + return self._normalize_v0_1_server(result) - return result - else: - if not data: - raise ValueError(f"Server '{server_id}' not found in registry") - return data + if not data: + raise ServerNotFoundError(server_name, self.registry_url) + return self._normalize_v0_1_server(data) + + def get_server_info(self, server_name: str) -> dict[str, Any]: + """Deprecated alias for :meth:`get_server`. + + Kept for one minor as a transitional shim; emits a + ``DeprecationWarning`` and forwards to ``get_server``. The + parameter is now interpreted as a server *name* (per the MCP + Registry v0.1 spec), not a UUID -- the legacy v0 ``/servers/{id}`` + endpoint no longer exists on spec-compliant registries. + """ + warnings.warn( + "SimpleRegistryClient.get_server_info(server_name) is deprecated; " + "use SimpleRegistryClient.get_server(server_name, version='latest'). " + "The parameter now means a server name per MCP Registry v0.1.", + DeprecationWarning, + stacklevel=2, + ) + return self.get_server(server_name) def get_server_by_name(self, name: str) -> dict[str, Any] | None: """Find a server by its name using the search API. @@ -340,21 +490,22 @@ def get_server_by_name(self, name: str) -> dict[str, Any] | None: for server in search_results: if server.get("name") == name: try: - return self.get_server_info(server["id"]) + return self.get_server(server["name"]) except ValueError: continue return None def find_server_by_reference(self, reference: str) -> dict[str, Any] | None: - """Find a server by exact name match or server ID. + """Find a server by exact name match. - This is an efficient lookup that uses the search API: - 1. Server ID (UUID format) - direct API call - 2. Server name - search API for exact match (automatically handles identifier extraction) + The legacy UUID strategy was removed because the MCP Registry v0.1 + spec keys per-server lookup on serverName, not UUID. Old UUID-style + references silently route through search and produce no match, + which is acceptable per design ratification. Args: - reference (str): Server reference (ID or exact name). + reference (str): Server reference (exact name or unqualified slug). Returns: Optional[Dict[str, Any]]: Server metadata dictionary or None if not found. @@ -362,16 +513,7 @@ def find_server_by_reference(self, reference: str) -> dict[str, Any] | None: Raises: requests.RequestException: If the registry API request fails. """ - # Strategy 1: Try as server ID first (direct lookup) - try: - # Check if it looks like a UUID (contains hyphens and is 36 chars) - if len(reference) == 36 and reference.count("-") == 4: - return self.get_server_info(reference) - except ValueError: - pass - - # Strategy 2: Use search API to find by name - # search_servers now handles extracting repository names internally + # Use search API to find by name search_results = self.search_servers(reference) # Pass 1: exact full-name match (prevents slug collisions) @@ -379,7 +521,7 @@ def find_server_by_reference(self, reference: str) -> dict[str, Any] | None: server_name = server.get("name", "") if server_name == reference: try: - return self.get_server_info(server["id"]) + return self.get_server(server_name) except ValueError: continue @@ -388,11 +530,11 @@ def find_server_by_reference(self, reference: str) -> dict[str, Any] | None: server_name = server.get("name", "") if self._is_server_match(reference, server_name): try: - return self.get_server_info(server["id"]) + return self.get_server(server_name) except ValueError: continue - # If not found by ID or exact name, server is not in registry + # If not found by name, server is not in registry return None def _extract_repository_name(self, reference: str) -> str: diff --git a/tests/integration/test_registry.py b/tests/integration/test_registry.py index 84372f95c..8e8af76dc 100644 --- a/tests/integration/test_registry.py +++ b/tests/integration/test_registry.py @@ -37,11 +37,17 @@ def safe_rmdir(path): class TestMCPRegistry: - """Test the MCP registry client with the demo registry.""" + """Test the MCP registry client end-to-end against the public GitHub MCP Registry. + + Previously targeted the legacy ``demo.registry.azure-mcp.net`` host + which only served the non-spec ``/v0/`` API (issue #1210). Retargeted + to ``api.mcp.github.com`` which is MCP Registry v0.1 spec-compliant. + """ def setup_method(self): """Set up test environment.""" - self.registry_client = SimpleRegistryClient("https://demo.registry.azure-mcp.net") + self.registry_url = "https://api.mcp.github.com" + self.registry_client = SimpleRegistryClient(self.registry_url) # Create a temporary directory for tests self.test_dir = tempfile.TemporaryDirectory() @@ -62,7 +68,7 @@ def teardown_method(self): # Leave the temp tree before unlinking it. Otherwise cwd can still # reference the directory inode and os.getcwd() raises FileNotFoundError - # on POSIX — breaking later tests on the same xdist worker. + # on POSIX -- breaking later tests on the same xdist worker. with contextlib.suppress(FileNotFoundError, OSError): os.chdir(tempfile.gettempdir()) @@ -76,41 +82,69 @@ def teardown_method(self): def test_list_servers(self): """Test listing servers from the registry.""" - servers, _ = self.registry_client.list_servers() + servers, _ = self.registry_client.list_servers(limit=5) assert isinstance(servers, list), "Server list should be a list" - assert len(servers) > 0, "Demo registry should have some servers" + assert len(servers) > 0, "Public registry should have some servers" - def test_get_server_info(self): - """Test getting server details for a specific server.""" - # Get the first server from the list - servers, _ = self.registry_client.list_servers() + def test_get_server(self): + """Test getting server details for a specific server (v0.1: keyed by name).""" + servers, _ = self.registry_client.list_servers(limit=5) if not servers: - pytest.skip("No servers available in the demo registry") + pytest.skip("No servers available in the registry") - server_id = servers[0]["id"] - server_info = self.registry_client.get_server_info(server_id) + server_name = servers[0]["name"] + server_info = self.registry_client.get_server(server_name) - assert server_info is not None, f"Server info for {server_id} should be retrievable" + assert server_info is not None, f"Server info for {server_name} should be retrievable" assert "name" in server_info, "Server info should include name" - assert "id" in server_info, "Server info should include id" + assert server_info["name"] == server_name def test_vscode_adapter_with_registry(self): """Test VSCode adapter with registry integration.""" - # Create a VSCode adapter - adapter = VSCodeClientAdapter("https://demo.registry.azure-mcp.net") + adapter = VSCodeClientAdapter(self.registry_url) - # Get a list of servers - servers, _ = self.registry_client.list_servers() + # Walk a small page to find a server whose packages map to a VSCode-supported + # transport (npm, pypi, docker). The first registry entry isn't guaranteed + # to be VSCode-compatible (e.g. uvx-only servers are skipped). + servers, _ = self.registry_client.list_servers(limit=20) if not servers: - pytest.skip("No servers available in the demo registry") - - # Configure the first server - server_id = servers[0]["id"] - result = adapter.configure_mcp_server(server_id) + pytest.skip("No servers available in the registry") + + configured = False + chosen_name = None + last_error: Exception | None = None + # Known unsupported-server failures: registry shapes the adapter + # actively skips (uvx-only, mcpb, etc.) raise ValueError or KeyError. + # Anything else is captured and re-raised at the end so a registry + # contract regression (e.g. v0.1 packages with `identifier` instead + # of `name`) cannot silently disguise itself as "no compatible + # server" -- which is the regression #1210 was filed against. + _expected_skip_errors = (ValueError, KeyError, TypeError) + for s in servers: + name = s.get("name") + if not name: + continue + try: + if adapter.configure_mcp_server(name) is True: + configured = True + chosen_name = name + break + except _expected_skip_errors: + continue + except Exception as exc: + last_error = exc + continue + + if not configured: + if last_error is not None: + raise AssertionError( + "VSCode adapter could not configure any of the first 20 registry " + "servers and the last failure was an unexpected exception (likely a " + "registry contract regression): " + f"{type(last_error).__name__}: {last_error}" + ) from last_error + pytest.skip("No VSCode-compatible server found in the first 20 registry entries") - assert result is True, f"Should be able to configure server {server_id}" - - # Check the generated configuration file config_path = os.path.join(self.test_dir.name, ".vscode", "mcp.json") assert os.path.exists(config_path), "Configuration file should be created" @@ -118,7 +152,5 @@ def test_vscode_adapter_with_registry(self): config = json.load(f) assert "servers" in config, "Config should have servers section" - - # The server name in the config will be the server_id unless a name was specified - assert server_id in config["servers"], f"Config should include {server_id}" - assert "type" in config["servers"][server_id], "Server config should have type" + assert chosen_name in config["servers"], f"Config should include {chosen_name}" + assert "type" in config["servers"][chosen_name], "Server config should have type" diff --git a/tests/integration/test_registry_client_integration.py b/tests/integration/test_registry_client_integration.py index 4980e1b60..2fd36efbf 100644 --- a/tests/integration/test_registry_client_integration.py +++ b/tests/integration/test_registry_client_integration.py @@ -1,6 +1,5 @@ -"""Integration tests for the MCP registry client with GitHub MCP Registry.""" +"""Integration tests for the MCP registry client with the GitHub MCP Registry (v0.1 spec).""" -import os # noqa: F401 import unittest import pytest @@ -12,14 +11,16 @@ class TestRegistryClientIntegration(unittest.TestCase): - """Integration test cases for the MCP registry client with the GitHub MCP Registry.""" + """Integration test cases against the live GitHub MCP Registry (https://api.mcp.github.com). + + These tests exercise the v0.1 spec endpoints end-to-end. They skip + cleanly when the registry is unreachable. + """ def setUp(self): """Set up test fixtures.""" - # Use the GitHub MCP Registry for integration tests self.client = SimpleRegistryClient("https://api.mcp.github.com") - # Skip tests if we can't reach the registry try: response = requests.head("https://api.mcp.github.com") # noqa: S113 response.raise_for_status() @@ -27,229 +28,91 @@ def setUp(self): self.skipTest("GitHub MCP Registry is not accessible") def test_list_servers(self): - """Test listing servers from the GitHub MCP Registry.""" + """list_servers returns spec-shaped server dicts with 'name'.""" try: - servers, next_cursor = self.client.list_servers() # noqa: RUF059 + servers, next_cursor = self.client.list_servers(limit=5) # noqa: RUF059 self.assertIsInstance(servers, list) - # We don't know exactly what servers will be in the demo registry, - # but we can check that the structure is correct if servers: + # v0.1 spec shape: 'name' is the stable identifier; 'id' may be absent. self.assertIn("name", servers[0]) - self.assertIn("id", servers[0]) except (requests.RequestException, ValueError) as e: self.skipTest(f"Could not list servers from GitHub MCP Registry: {e}") def test_search_servers(self): - """Test searching for servers in the registry using the search API.""" + """search_servers hits the v0.1 ?search= query and returns matches.""" try: - # Test search with a common term like "github" results = self.client.search_servers("github") - - # We should find some results self.assertGreater(len(results), 0, "Search should return at least some results") - # Each result should have basic server structure for server in results: self.assertIn("name", server) - self.assertIn("id", server) self.assertIn("description", server) - # Test that search actually filters (empty search should return fewer or different results) - # Note: We can't guarantee behavior of empty searches, so we test with a specific term - specific_results = self.client.search_servers("universal") - # The results should be a list (could be empty if no matches) + specific_results = self.client.search_servers("nonexistent-xyz-123-needle") self.assertIsInstance(specific_results, list) - except (requests.RequestException, ValueError) as e: self.skipTest(f"Could not search servers in registry: {e}") - def test_get_server_info(self): - """Test getting server information from the GitHub MCP Registry.""" + def test_get_server_uses_v0_1_versions_latest(self): + """get_server resolves a name via /v0.1/servers/{name}/versions/latest.""" try: - # First, get all servers to find one to get info about - all_servers, _ = self.client.list_servers() + all_servers, _ = self.client.list_servers(limit=5) if not all_servers: - self.skipTest("No servers found in GitHub MCP Registry to get info about") + self.skipTest("No servers found in GitHub MCP Registry to look up") - # Get info about the first server - server_id = all_servers[0]["id"] - server_info = self.client.get_server_info(server_id) + server_name = all_servers[0]["name"] + server_info = self.client.get_server(server_name) - # Check that we got the expected server info - self.assertIn("name", server_info) - self.assertEqual(server_info["id"], server_id) + self.assertEqual(server_info["name"], server_name) self.assertIn("description", server_info) + self.assertIn("version", server_info) - # Check for version_detail - self.assertIn("version_detail", server_info) - if "version_detail" in server_info: - self.assertIn("version", server_info["version_detail"]) - - # Check for packages if available if server_info.get("packages"): pkg = server_info["packages"][0] - self.assertIn("name", pkg) + # v0.1 spec uses 'identifier' (was 'name' in legacy v0). + self.assertTrue("identifier" in pkg or "name" in pkg) self.assertIn("version", pkg) except (requests.RequestException, ValueError) as e: self.skipTest(f"Could not get server info from GitHub MCP Registry: {e}") def test_get_server_by_name(self): - """Test finding a server by name.""" + """get_server_by_name finds a known server and returns None for missing names.""" try: - # First, get all servers to find one to look up - all_servers, _ = self.client.list_servers() + all_servers, _ = self.client.list_servers(limit=5) if not all_servers: self.skipTest("No servers found in GitHub MCP Registry to look up") - # Try to find the first server by name server_name = all_servers[0]["name"] found_server = self.client.get_server_by_name(server_name) - # Check that we found the expected server self.assertIsNotNone(found_server, "Server should be found by name") self.assertEqual(found_server["name"], server_name) - # Try with a non-existent name - non_existent = self.client.get_server_by_name("non-existent-server-name-12345") + non_existent = self.client.get_server_by_name( + "io.github.nonexistent/definitely-not-a-real-server-12345" + ) self.assertIsNone(non_existent, "Non-existent server should return None") except (requests.RequestException, ValueError) as e: self.skipTest(f"Could not find server by name in GitHub MCP Registry: {e}") - def test_specific_real_servers(self): - """Test integration with specific real servers from the GitHub MCP Registry.""" - # Test specific server IDs from the GitHub MCP Registry - universal_server_id = "cb84de60-6710-40eb-8cb3-a350ce27c34e" # Universal MCP (pip runtime) - docker_server_id = "52dd9765-6aea-476a-9338-5ffe1ddbefc5" # it-tools (docker runtime) - npx_server_id = "f3432bd2-9c05-4b27-b0f4-f4e7a83dbf66" # deepseek-mcp-server (npx runtime) - - # Set to collect different runtime types we encounter - runtime_types = set() - - # Test the Universal MCP server (pip runtime) - try: - universal_server = self.client.get_server_info(universal_server_id) - - # Validate basic server information - self.assertEqual(universal_server["id"], universal_server_id) - self.assertIn("name", universal_server) - self.assertIn("description", universal_server) - - # Validate repository information - self.assertIn("repository", universal_server) - self.assertIn("url", universal_server["repository"]) - self.assertIn("source", universal_server["repository"]) - - # Validate version details - self.assertIn("version_detail", universal_server) - self.assertIn("version", universal_server["version_detail"]) - - # Validate it has package information - self.assertIn("packages", universal_server) - self.assertGreater(len(universal_server["packages"]), 0) - - # Validate package details - package = universal_server["packages"][0] - self.assertIn("name", package) - self.assertIn("version", package) - - if "runtime_hint" in package: - runtime_types.add(package["runtime_hint"]) - - # Test finding by name - universal_name = universal_server["name"] - found_by_name = self.client.get_server_by_name(universal_name) - self.assertIsNotNone(found_by_name) - self.assertEqual(found_by_name["id"], universal_server_id) - except (requests.RequestException, ValueError) as e: - self.skipTest(f"Could not test Universal MCP server: {e}") - - # Test the Docker server (docker runtime) - try: - docker_server = self.client.get_server_info(docker_server_id) - - # Validate basic server information - self.assertEqual(docker_server["id"], docker_server_id) - self.assertIn("name", docker_server) - self.assertIn("description", docker_server) - - # Validate repository information - self.assertIn("repository", docker_server) - self.assertIn("url", docker_server["repository"]) - - # Validate it has package information if available - if docker_server.get("packages"): - package = docker_server["packages"][0] - self.assertIn("name", package) - self.assertIn("version", package) - - if "runtime_hint" in package: - runtime_types.add(package["runtime_hint"]) - except (requests.RequestException, ValueError) as e: - self.skipTest(f"Could not test Docker MCP server: {e}") + def test_get_server_url_encodes_slash(self): + """A serverName containing '/' is URL-encoded to %2F when sent to the registry. - # Test the NPX server (npx runtime) + Regression trap for issue #1210: the spec requires URL-encoding of + the path segment so registries can route the lookup correctly. + """ try: - npx_server = self.client.get_server_info(npx_server_id) - - # Validate basic server information - self.assertEqual(npx_server["id"], npx_server_id) - self.assertIn("name", npx_server) - self.assertIn("description", npx_server) - - # Validate it has package information if available - if npx_server.get("packages"): - package = npx_server["packages"][0] - self.assertIn("name", package) - self.assertIn("version", package) - - if "runtime_hint" in package: - runtime_types.add(package["runtime_hint"]) - except (requests.RequestException, ValueError) as e: - self.skipTest(f"Could not test NPX MCP server: {e}") - - # Try to find a server with another runtime type for more diversity - try: - # Search for servers with different runtime types - servers, _ = self.client.list_servers(limit=30) - - for server in servers: - server_id = server["id"] - if server_id not in [universal_server_id, docker_server_id, npx_server_id]: - try: - server_info = self.client.get_server_info(server_id) - - if server_info.get("packages"): - for package in server_info["packages"]: - if ( - "runtime_hint" in package - and package["runtime_hint"] not in runtime_types - ): - runtime_types.add(package["runtime_hint"]) - - # Validate we can get basic info for this server type - self.assertIn("name", server_info) - self.assertIn("description", server_info) - self.assertIn("id", server_info) - - # If we found at least 3 different runtime types, we've validated enough diversity - if len(runtime_types) >= 3: - break - except (requests.RequestException, ValueError): - # Skip servers that can't be accessed - continue - - # If we found at least 3 different runtime types, we've validated enough diversity - if len(runtime_types) >= 3: - break - - # We should have found at least 2 different runtime types from our test servers - self.assertGreaterEqual( - len(runtime_types), - 2, - f"Expected to find at least 2 different runtime types, found: {runtime_types}", - ) + servers, _ = self.client.list_servers(limit=20) + namespaced = next((s for s in servers if "/" in s.get("name", "")), None) + if namespaced is None: + self.skipTest("No namespaced servers found to validate URL encoding") + + # If get_server returns successfully, the URL was correctly encoded + # (a raw slash would 404 because it would be interpreted as a path separator). + info = self.client.get_server(namespaced["name"]) + self.assertEqual(info["name"], namespaced["name"]) except (requests.RequestException, ValueError) as e: - self.skipTest(f"Could not test servers with different runtime types: {e}") + self.skipTest(f"Could not validate URL encoding against live registry: {e}") if __name__ == "__main__": diff --git a/tests/unit/test_registry_client.py b/tests/unit/test_registry_client.py index 0fc7beb2a..5ed4d3c60 100644 --- a/tests/unit/test_registry_client.py +++ b/tests/unit/test_registry_client.py @@ -2,7 +2,9 @@ import os import unittest +import warnings from unittest import mock +from urllib.parse import urlparse import requests @@ -19,168 +21,268 @@ def setUp(self): @mock.patch("requests.Session.get") def test_list_servers(self, mock_get): - """Test listing servers from the registry.""" - # Mock response + """Test listing servers from the registry under the v0.1 spec.""" mock_response = mock.Mock() mock_response.json.return_value = { "servers": [ - { - "id": "123e4567-e89b-12d3-a456-426614174000", - "name": "server1", - "description": "Description 1", - }, - { - "id": "223e4567-e89b-12d3-a456-426614174000", - "name": "server2", - "description": "Description 2", - }, + {"server": {"name": "io.github.foo/server1", "description": "Description 1"}}, + {"server": {"name": "io.github.foo/server2", "description": "Description 2"}}, ], - "metadata": {"next_cursor": "next-page-token", "count": 2}, + "metadata": {"nextCursor": "next-page-token", "count": 2}, } mock_response.raise_for_status.return_value = None mock_get.return_value = mock_response - # Call the method servers, next_cursor = self.client.list_servers() - # Assertions self.assertEqual(len(servers), 2) - self.assertEqual(servers[0]["name"], "server1") - self.assertEqual(servers[1]["name"], "server2") + self.assertEqual(servers[0]["name"], "io.github.foo/server1") + self.assertEqual(servers[1]["name"], "io.github.foo/server2") self.assertEqual(next_cursor, "next-page-token") mock_get.assert_called_once_with( - f"{self.client.registry_url}/v0/servers", + f"{self.client.registry_url}/v0.1/servers", params={"limit": 100}, timeout=self.client._timeout, ) @mock.patch("requests.Session.get") def test_list_servers_with_pagination(self, mock_get): - """Test listing servers with pagination parameters.""" - # Mock response + """Test listing servers with pagination parameters under the v0.1 spec.""" mock_response = mock.Mock() mock_response.json.return_value = {"servers": [], "metadata": {}} mock_response.raise_for_status.return_value = None mock_get.return_value = mock_response - # Call the method with pagination self.client.list_servers(limit=10, cursor="page-token") - # Assertions mock_get.assert_called_once_with( - f"{self.client.registry_url}/v0/servers", + f"{self.client.registry_url}/v0.1/servers", params={"limit": 10, "cursor": "page-token"}, timeout=self.client._timeout, ) + @mock.patch("requests.Session.get") + def test_list_servers_reads_nextCursor_camelCase(self, mock_get): + """Regression trap (#1210): metadata.nextCursor is the spec key.""" + mock_response = mock.Mock() + mock_response.json.return_value = { + "servers": [], + "metadata": {"nextCursor": "spec-cursor"}, + } + mock_response.raise_for_status.return_value = None + mock_get.return_value = mock_response + + _, next_cursor = self.client.list_servers() + self.assertEqual(next_cursor, "spec-cursor") + + @mock.patch("requests.Session.get") + def test_list_servers_uses_v0_1_endpoint(self, mock_get): + """Regression trap (#1210): URL must be /v0.1/, never /v0/.""" + mock_response = mock.Mock() + mock_response.json.return_value = {"servers": [], "metadata": {}} + mock_response.raise_for_status.return_value = None + mock_get.return_value = mock_response + + self.client.list_servers() + called_url = mock_get.call_args[0][0] + parsed_path = urlparse(called_url).path + self.assertEqual(parsed_path, "/v0.1/servers") + self.assertNotEqual(parsed_path, "/v0/servers") + @mock.patch("requests.Session.get") def test_search_servers(self, mock_get): - """Test searching for servers in the registry using API search endpoint.""" - # Mock response + """Test searching for servers under the v0.1 spec (?search= on /v0.1/servers).""" mock_response = mock.Mock() mock_response.json.return_value = { "servers": [ - {"server": {"name": "test-server", "description": "Test description"}}, - {"server": {"name": "server2", "description": "Another test"}}, + { + "server": { + "name": "io.github.foo/test-server", + "description": "Test description", + } + }, + {"server": {"name": "io.github.foo/server2", "description": "Another test"}}, ] } mock_response.raise_for_status.return_value = None mock_get.return_value = mock_response - # Call the method with a search query results = self.client.search_servers("test") - # Assertions mock_get.assert_called_once_with( - f"{self.client.registry_url}/v0/servers/search", - params={"q": "test"}, + f"{self.client.registry_url}/v0.1/servers", + params={"search": "test"}, timeout=self.client._timeout, ) self.assertEqual(len(results), 2) - self.assertEqual(results[0]["name"], "test-server") - self.assertEqual(results[1]["name"], "server2") + self.assertEqual(results[0]["name"], "io.github.foo/test-server") + self.assertEqual(results[1]["name"], "io.github.foo/server2") @mock.patch("requests.Session.get") - def test_get_server_info(self, mock_get): - """Test getting server information from the registry.""" - # Mock response + def test_search_servers_passes_full_reference(self, mock_get): + """search_servers passes the full reference (no slug pre-trim) to the spec ?search=""" + mock_response = mock.Mock() + mock_response.json.return_value = {"servers": []} + mock_response.raise_for_status.return_value = None + mock_get.return_value = mock_response + + self.client.search_servers("io.github.foo/bar") + + mock_get.assert_called_once_with( + f"{self.client.registry_url}/v0.1/servers", + params={"search": "io.github.foo/bar"}, + timeout=self.client._timeout, + ) + + @mock.patch("requests.Session.get") + def test_get_server(self, mock_get): + """Test getting server info under the v0.1 spec.""" mock_response = mock.Mock() server_data = { - "id": "123e4567-e89b-12d3-a456-426614174000", - "name": "test-server", - "description": "Test server description", - "repository": { - "url": f"https://{github_host.default_host()}/test/test-server", - "source": "github", - "id": "12345", - }, - "version_detail": { + "server": { + "name": "io.github.foo/test-server", + "description": "Test server description", + "repository": { + "url": f"https://{github_host.default_host()}/foo/test-server", + "source": "github", + "id": "12345", + }, "version": "1.0.0", - "release_date": "2025-05-16T19:13:21Z", - "is_latest": True, - }, - "package_canonical": "npm", - "packages": [ - { - "registry_name": "npm", - "name": "test-package", - "version": "1.0.0", - "runtime_hint": "npx", - } - ], + "packages": [ + { + "registry_name": "npm", + "name": "test-package", + "version": "1.0.0", + "runtime_hint": "npx", + } + ], + } } mock_response.json.return_value = server_data mock_response.raise_for_status.return_value = None mock_get.return_value = mock_response - # Call the method - server_info = self.client.get_server_info("123e4567-e89b-12d3-a456-426614174000") + server_info = self.client.get_server("io.github.foo/test-server") - # Assertions - self.assertEqual(server_info["name"], "test-server") - self.assertEqual(server_info["version_detail"]["version"], "1.0.0") + self.assertEqual(server_info["name"], "io.github.foo/test-server") + self.assertEqual(server_info["version"], "1.0.0") self.assertEqual(server_info["packages"][0]["name"], "test-package") mock_get.assert_called_once_with( - f"{self.client.registry_url}/v0/servers/123e4567-e89b-12d3-a456-426614174000", + f"{self.client.registry_url}/v0.1/servers/io.github.foo%2Ftest-server/versions/latest", timeout=self.client._timeout, ) + @mock.patch("requests.Session.get") + def test_get_server_url_encodes_slash_in_name(self, mock_get): + """Regression trap (#1210): slash in serverName must be URL-encoded as %2F.""" + mock_response = mock.Mock() + mock_response.json.return_value = {"server": {"name": "io.github.github/github-mcp-server"}} + mock_response.raise_for_status.return_value = None + mock_get.return_value = mock_response + + self.client.get_server("io.github.github/github-mcp-server") + + called_url = mock_get.call_args[0][0] + parsed_path = urlparse(called_url).path + # serverName slash must be encoded as %2F so the registry routes the + # name as a single path segment, not as nested servers//. + self.assertEqual( + parsed_path, + "/v0.1/servers/io.github.github%2Fgithub-mcp-server/versions/latest", + ) + + def test_get_server_rejects_invalid_name_shape(self): + """get_server rejects names that don't match the MCP spec shape.""" + for bad in ["", "../etc/passwd", "https://evil/", "name with space", "a/b/c"]: + with self.subTest(name=bad): + with self.assertRaises(ValueError): + self.client.get_server(bad) + + @mock.patch("requests.Session.get") + def test_get_server_404_raises_server_not_found(self, mock_get): + """get_server wraps 404s in ServerNotFoundError carrying the registry URL.""" + from apm_cli.registry.client import ServerNotFoundError + + mock_response = mock.Mock() + mock_response.status_code = 404 + http_error = requests.HTTPError("404 Not Found") + http_error.response = mock_response + mock_response.raise_for_status.side_effect = http_error + mock_get.return_value = mock_response + + with self.assertRaises(ServerNotFoundError) as cm: + self.client.get_server("io.github.foo/missing") + msg = str(cm.exception) + self.assertIn("io.github.foo/missing", msg) + # Validate the registry URL appears in the error by parsing it out of + # the message rather than substring-matching the raw URL (CodeQL rule + # `py/incomplete-url-substring-sanitization`). Find the URL token that + # carries a scheme and assert hostname / scheme equality. + registry_parsed = urlparse(self.client.registry_url) + url_tokens = [tok.strip(".,;'\")(") for tok in msg.split() if "://" in tok] + parsed_tokens = [urlparse(tok) for tok in url_tokens] + self.assertTrue( + any( + p.scheme == registry_parsed.scheme and p.hostname == registry_parsed.hostname + for p in parsed_tokens + ), + f"Registry URL not present in error message: {msg!r}", + ) + # ServerNotFoundError must be a ValueError subclass for legacy callers + self.assertIsInstance(cm.exception, ValueError) + + def test_get_server_info_is_deprecated_shim(self): + """get_server_info is a one-minor deprecation shim that forwards to get_server.""" + with mock.patch.object(self.client, "get_server") as mock_get_server: + mock_get_server.return_value = {"name": "io.github.foo/bar"} + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + result = self.client.get_server_info("io.github.foo/bar") + self.assertEqual(result, {"name": "io.github.foo/bar"}) + mock_get_server.assert_called_once_with("io.github.foo/bar") + self.assertTrue( + any(issubclass(w.category, DeprecationWarning) for w in caught), + f"Expected DeprecationWarning, got {[w.category for w in caught]}", + ) + @mock.patch("apm_cli.registry.client.SimpleRegistryClient.search_servers") - @mock.patch("apm_cli.registry.client.SimpleRegistryClient.get_server_info") - def test_get_server_by_name(self, mock_get_server_info, mock_search_servers): - """Test finding a server by name using search API.""" - # Mock search_servers + @mock.patch("apm_cli.registry.client.SimpleRegistryClient.get_server") + def test_get_server_by_name(self, mock_get_server, mock_search_servers): + """Test finding a server by name using search API (v0.1 shape: name, no id).""" mock_search_servers.return_value = [ - {"id": "123e4567-e89b-12d3-a456-426614174000", "name": "test-server"}, - {"id": "223e4567-e89b-12d3-a456-426614174000", "name": "other-server"}, + {"name": "test-server"}, + {"name": "other-server"}, ] + server_data = {"name": "test-server", "description": "Test server"} + mock_get_server.return_value = server_data - # Mock get_server_info - server_data = { - "id": "123e4567-e89b-12d3-a456-426614174000", - "name": "test-server", - "description": "Test server", - } - mock_get_server_info.return_value = server_data - - # Test finding server using search result = self.client.get_server_by_name("test-server") - # Assertions self.assertEqual(result, server_data) mock_search_servers.assert_called_once_with("test-server") - mock_get_server_info.assert_called_once_with("123e4567-e89b-12d3-a456-426614174000") + mock_get_server.assert_called_once_with("test-server") # Reset mocks for non-existent test - mock_get_server_info.reset_mock() + mock_get_server.reset_mock() mock_search_servers.reset_mock() - mock_search_servers.return_value = [] # No search results + mock_search_servers.return_value = [] - # Test non-existent server result = self.client.get_server_by_name("non-existent") self.assertIsNone(result) mock_search_servers.assert_called_once_with("non-existent") - mock_get_server_info.assert_not_called() + mock_get_server.assert_not_called() + + def test_get_server_by_name_does_not_require_top_level_id(self): + """Regression trap (#1210): lookup chain must work without top-level 'id'.""" + with ( + mock.patch.object(self.client, "search_servers") as mock_search, + mock.patch.object(self.client, "get_server") as mock_get, + ): + mock_search.return_value = [{"name": "io.github.foo/bar"}] # NO 'id' key + mock_get.return_value = {"name": "io.github.foo/bar"} + result = self.client.get_server_by_name("io.github.foo/bar") + self.assertEqual(result, {"name": "io.github.foo/bar"}) + mock_get.assert_called_once_with("io.github.foo/bar") @mock.patch.dict(os.environ, {"MCP_REGISTRY_URL": "https://custom-registry.example.com"}) def test_environment_variable_override(self): @@ -192,125 +294,66 @@ def test_environment_variable_override(self): client = SimpleRegistryClient("https://explicit-url.example.com") self.assertEqual(client.registry_url, "https://explicit-url.example.com") - @mock.patch("apm_cli.registry.client.SimpleRegistryClient.get_server_info") - def test_find_server_by_reference_uuid(self, mock_get_server_info): - """Test finding a server by UUID reference.""" - # Mock server data - server_data = { - "id": "123e4567-e89b-12d3-a456-426614174000", - "name": "test-server", - "description": "Test server", - } - mock_get_server_info.return_value = server_data - - # Call the method with UUID - result = self.client.find_server_by_reference("123e4567-e89b-12d3-a456-426614174000") - - # Assertions - self.assertEqual(result, server_data) - mock_get_server_info.assert_called_once_with("123e4567-e89b-12d3-a456-426614174000") - @mock.patch("apm_cli.registry.client.SimpleRegistryClient.search_servers") - @mock.patch("apm_cli.registry.client.SimpleRegistryClient.get_server_info") - def test_find_server_by_reference_uuid_not_found( - self, mock_get_server_info, mock_search_servers - ): - """Test finding a server by UUID that doesn't exist. - - When the UUID lookup raises ValueError, ``find_server_by_reference`` falls - through to ``search_servers``; that fallback must be mocked so the test - never hits the real registry (the Windows CI runner cannot resolve - ``api.mcp.github.com`` and the call would raise ``socket.gaierror``). - """ - # Mock get_server_info to raise ValueError so the UUID branch fails. - mock_get_server_info.side_effect = ValueError("Server not found") - # Mock the fallback search to return no hits (the "not found" case). + def test_find_server_by_reference_uuid_input_returns_none(self, mock_search_servers): + """The legacy UUID strategy is removed; UUID-shaped refs route to search and miss.""" mock_search_servers.return_value = [] - - # Call the method with UUID result = self.client.find_server_by_reference("123e4567-e89b-12d3-a456-426614174000") - - # Should return None when server not found self.assertIsNone(result) - mock_get_server_info.assert_called_once_with("123e4567-e89b-12d3-a456-426614174000") mock_search_servers.assert_called_once_with("123e4567-e89b-12d3-a456-426614174000") - @mock.patch("apm_cli.registry.client.SimpleRegistryClient.get_server_info") + @mock.patch("apm_cli.registry.client.SimpleRegistryClient.get_server") @mock.patch("apm_cli.registry.client.SimpleRegistryClient.search_servers") - def test_find_server_by_reference_name_match(self, mock_search_servers, mock_get_server_info): - """Test finding a server by exact name match.""" - # Mock search_servers + def test_find_server_by_reference_name_match(self, mock_search_servers, mock_get_server): + """Test finding a server by exact name match (v0.1 shape).""" mock_search_servers.return_value = [ - {"id": "123e4567-e89b-12d3-a456-426614174000", "name": "io.github.owner/repo-name"}, - {"id": "223e4567-e89b-12d3-a456-426614174000", "name": "other-server"}, + {"name": "io.github.owner/repo-name"}, + {"name": "other-server"}, ] + server_data = {"name": "io.github.owner/repo-name", "description": "Test server"} + mock_get_server.return_value = server_data - # Mock get_server_info - server_data = { - "id": "123e4567-e89b-12d3-a456-426614174000", - "name": "io.github.owner/repo-name", - "description": "Test server", - } - mock_get_server_info.return_value = server_data - - # Call the method with exact name result = self.client.find_server_by_reference("io.github.owner/repo-name") - # Assertions self.assertEqual(result, server_data) mock_search_servers.assert_called_once_with("io.github.owner/repo-name") - mock_get_server_info.assert_called_once_with("123e4567-e89b-12d3-a456-426614174000") + mock_get_server.assert_called_once_with("io.github.owner/repo-name") @mock.patch("apm_cli.registry.client.SimpleRegistryClient.search_servers") def test_find_server_by_reference_name_not_found(self, mock_search_servers): """Test finding a server by name that doesn't exist in registry.""" - # Mock search_servers with no matching names mock_search_servers.return_value = [ - { - "id": "123e4567-e89b-12d3-a456-426614174000", - "name": "io.github.owner/different-repo", - }, - {"id": "223e4567-e89b-12d3-a456-426614174000", "name": "other-server"}, + {"name": "io.github.owner/different-repo"}, + {"name": "other-server"}, ] - # Call the method with non-existent name result = self.client.find_server_by_reference("ghcr.io/github/github-mcp-server") - # Should return None when server not found self.assertIsNone(result) mock_search_servers.assert_called_once_with("ghcr.io/github/github-mcp-server") - @mock.patch("apm_cli.registry.client.SimpleRegistryClient.get_server_info") + @mock.patch("apm_cli.registry.client.SimpleRegistryClient.get_server") @mock.patch("apm_cli.registry.client.SimpleRegistryClient.search_servers") - def test_find_server_by_reference_name_match_get_server_info_fails( - self, mock_search_servers, mock_get_server_info + def test_find_server_by_reference_name_match_get_server_fails( + self, mock_search_servers, mock_get_server ): - """Test finding a server by name when get_server_info raises ValueError (stale ID).""" - # Mock search_servers - mock_search_servers.return_value = [ - {"id": "123e4567-e89b-12d3-a456-426614174000", "name": "test-server"} - ] + """When get_server raises ValueError (e.g. ServerNotFoundError), return None.""" + mock_search_servers.return_value = [{"name": "test-server"}] + mock_get_server.side_effect = ValueError("Server not found") - # Mock get_server_info to fail with ValueError (server not found by ID) - mock_get_server_info.side_effect = ValueError("Server not found") - - # Should return None when get_server_info fails with ValueError result = self.client.find_server_by_reference("test-server") self.assertIsNone(result) mock_search_servers.assert_called_once_with("test-server") - @mock.patch("apm_cli.registry.client.SimpleRegistryClient.get_server_info") + @mock.patch("apm_cli.registry.client.SimpleRegistryClient.get_server") @mock.patch("apm_cli.registry.client.SimpleRegistryClient.search_servers") def test_find_server_by_reference_name_match_network_error_propagates( - self, mock_search_servers, mock_get_server_info + self, mock_search_servers, mock_get_server ): - """Test that network errors in get_server_info propagate to the caller.""" - mock_search_servers.return_value = [ - {"id": "123e4567-e89b-12d3-a456-426614174000", "name": "test-server"} - ] - - mock_get_server_info.side_effect = requests.ConnectionError("Network error") + """Test that network errors in get_server propagate to the caller.""" + mock_search_servers.return_value = [{"name": "test-server"}] + mock_get_server.side_effect = requests.ConnectionError("Network error") with self.assertRaises(requests.ConnectionError): self.client.find_server_by_reference("test-server") @@ -318,16 +361,14 @@ def test_find_server_by_reference_name_match_network_error_propagates( @mock.patch("apm_cli.registry.client.SimpleRegistryClient.search_servers") def test_find_server_by_reference_invalid_format(self, mock_search_servers): """Test finding a server with various invalid/edge case formats.""" - # Mock search_servers with no matches mock_search_servers.return_value = [] - # Test various formats that should not match test_cases = [ - "", # Empty string - "short", # Too short to be UUID - "123e4567-e89b-12d3-a456-426614174000-extra", # Too long to be UUID - "not-a-uuid-but-36-chars-long-string", # 36 chars but wrong format - "registry.io/very/long/path/name", # Container-like reference + "", + "short", + "123e4567-e89b-12d3-a456-426614174000-extra", + "not-a-uuid-but-36-chars-long-string", + "registry.io/very/long/path/name", ] for test_case in test_cases: @@ -335,39 +376,36 @@ def test_find_server_by_reference_invalid_format(self, mock_search_servers): result = self.client.find_server_by_reference(test_case) self.assertIsNone(result) - @mock.patch("apm_cli.registry.client.SimpleRegistryClient.get_server_info") + @mock.patch("apm_cli.registry.client.SimpleRegistryClient.get_server") @mock.patch("apm_cli.registry.client.SimpleRegistryClient.search_servers") - def test_find_server_by_reference_no_slug_collision( - self, mock_search_servers, mock_get_server_info - ): + def test_find_server_by_reference_no_slug_collision(self, mock_search_servers, mock_get_server): """Test that qualified names don't collide on shared slugs (bug #165).""" - # Registry returns multiple servers sharing the slug 'mcp' mock_search_servers.return_value = [ - {"id": "aaa", "name": "com.supabase/mcp"}, - {"id": "bbb", "name": "microsoftdocs/mcp"}, + {"name": "com.supabase/mcp"}, + {"name": "microsoftdocs/mcp"}, ] - server_data = {"id": "bbb", "name": "microsoftdocs/mcp", "description": "MS Docs"} - mock_get_server_info.return_value = server_data + server_data = {"name": "microsoftdocs/mcp", "description": "MS Docs"} + mock_get_server.return_value = server_data result = self.client.find_server_by_reference("microsoftdocs/mcp") self.assertEqual(result, server_data) - mock_get_server_info.assert_called_once_with("bbb") + mock_get_server.assert_called_once_with("microsoftdocs/mcp") - @mock.patch("apm_cli.registry.client.SimpleRegistryClient.get_server_info") + @mock.patch("apm_cli.registry.client.SimpleRegistryClient.get_server") @mock.patch("apm_cli.registry.client.SimpleRegistryClient.search_servers") def test_find_server_by_reference_qualified_no_match( - self, mock_search_servers, mock_get_server_info + self, mock_search_servers, mock_get_server ): """Test that a qualified name with no exact match returns None.""" mock_search_servers.return_value = [ - {"id": "aaa", "name": "com.supabase/mcp"}, + {"name": "com.supabase/mcp"}, ] result = self.client.find_server_by_reference("microsoftdocs/mcp") self.assertIsNone(result) - mock_get_server_info.assert_not_called() + mock_get_server.assert_not_called() def test_is_server_match_qualified_prevents_collision(self): """Test _is_server_match rejects different namespaces with same slug.""" @@ -489,6 +527,134 @@ def test_env_var_invalid_rejected(self): SimpleRegistryClient() self.assertIn("MCP_REGISTRY_URL", str(cm.exception)) + def test_userinfo_stripped_from_registry_url(self): + """SimpleRegistryClient must strip user:pass@ from the stored URL. + + Regression trap for the credential-leak path: if userinfo survives + into ``self.registry_url``, ``ServerNotFoundError`` interpolates it + into terminal output and CI logs. + """ + c = SimpleRegistryClient("https://token:x-oauth@registry.corp.example.com/") + parsed = urlparse(c.registry_url) + self.assertEqual(parsed.scheme, "https") + self.assertEqual(parsed.hostname, "registry.corp.example.com") + self.assertIsNone(parsed.username) + self.assertIsNone(parsed.password) + self.assertEqual(c.registry_url, "https://registry.corp.example.com") + + def test_userinfo_stripped_preserves_explicit_port(self): + c = SimpleRegistryClient("https://user:pass@registry.corp.example.com:8443/") + parsed = urlparse(c.registry_url) + self.assertEqual(parsed.hostname, "registry.corp.example.com") + self.assertEqual(parsed.port, 8443) + self.assertIsNone(parsed.username) + self.assertIsNone(parsed.password) + + +class TestNormalizeV01Package(unittest.TestCase): + """Unit tests for the v0.1 -> snake_case package shape normalizer. + + Regression trap for #1210: registry returns camelCase keys per the + v0.1 spec, but adapters in ``src/apm_cli/adapters/client/`` consume + snake_case keys (``name``, ``runtime_hint``, ``package_arguments``, + ...). Without the boundary normalizer, ``apm install`` produced + ``npx -y None`` because the resolved package dict had no ``name``. + """ + + def test_normalize_v0_1_package_backfills_all_aliases(self): + """Every v0.1 camelCase alias backfills its snake_case counterpart.""" + v0_1_package = { + "identifier": "@modelcontextprotocol/server-fetch", + "registryType": "npm", + "registryBaseUrl": "https://registry.npmjs.org", + "runtimeHint": "npx", + "packageArguments": [{"value": "-y"}], + "runtimeArguments": [{"value": "--no-install"}], + "environmentVariables": [{"name": "DEBUG", "value": "1"}], + "version": "1.0.0", + } + normalized = SimpleRegistryClient._normalize_v0_1_package(v0_1_package) + + self.assertEqual(normalized["name"], "@modelcontextprotocol/server-fetch") + self.assertEqual(normalized["registry_name"], "npm") + self.assertEqual(normalized["registry_base_url"], "https://registry.npmjs.org") + self.assertEqual(normalized["runtime_hint"], "npx") + self.assertEqual(normalized["package_arguments"], [{"value": "-y"}]) + self.assertEqual(normalized["runtime_arguments"], [{"value": "--no-install"}]) + self.assertEqual(normalized["environment_variables"], [{"name": "DEBUG", "value": "1"}]) + self.assertEqual(normalized["version"], "1.0.0") + + # camelCase keys are preserved (one-way bridge, not a rewrite). + self.assertEqual(normalized["identifier"], "@modelcontextprotocol/server-fetch") + self.assertEqual(normalized["runtimeHint"], "npx") + + def test_normalize_v0_1_package_preserves_existing_legacy_keys(self): + """When both camelCase and snake_case are present, snake_case wins.""" + package = { + "identifier": "v0.1-name", + "name": "legacy-name", + "runtimeHint": "v0.1-hint", + "runtime_hint": "legacy-hint", + } + normalized = SimpleRegistryClient._normalize_v0_1_package(package) + self.assertEqual(normalized["name"], "legacy-name") + self.assertEqual(normalized["runtime_hint"], "legacy-hint") + + def test_normalize_v0_1_package_does_not_mutate_input(self): + """Normalizer returns a shallow copy; the input dict is unchanged.""" + package = {"identifier": "foo", "runtimeHint": "npx"} + SimpleRegistryClient._normalize_v0_1_package(package) + self.assertNotIn("name", package) + self.assertNotIn("runtime_hint", package) + + def test_normalize_v0_1_package_handles_partial_aliases(self): + """Aliases backfill independently; missing v0.1 keys are no-ops.""" + package = {"identifier": "foo", "version": "1.0.0"} + normalized = SimpleRegistryClient._normalize_v0_1_package(package) + self.assertEqual(normalized["name"], "foo") + self.assertNotIn("runtime_hint", normalized) + self.assertNotIn("registry_name", normalized) + + def test_normalize_v0_1_package_returns_non_dict_unchanged(self): + self.assertIsNone(SimpleRegistryClient._normalize_v0_1_package(None)) + self.assertEqual(SimpleRegistryClient._normalize_v0_1_package("not-a-dict"), "not-a-dict") + + def test_normalize_v0_1_server_normalizes_packages_list(self): + """Server-level normalizer applies package normalization to each entry.""" + server = { + "name": "io.github.foo/bar", + "version": "1.0.0", + "packages": [ + {"identifier": "pkg-a", "runtimeHint": "npx"}, + {"identifier": "pkg-b", "registryType": "pypi"}, + ], + } + normalized = SimpleRegistryClient._normalize_v0_1_server(server) + self.assertEqual(normalized["packages"][0]["name"], "pkg-a") + self.assertEqual(normalized["packages"][0]["runtime_hint"], "npx") + self.assertEqual(normalized["packages"][1]["name"], "pkg-b") + self.assertEqual(normalized["packages"][1]["registry_name"], "pypi") + + def test_normalize_v0_1_server_does_not_mutate_input(self): + """Server-level normalizer matches sibling copy semantics (no mutation).""" + server = { + "name": "io.github.foo/bar", + "packages": [{"identifier": "pkg-a", "runtimeHint": "npx"}], + } + SimpleRegistryClient._normalize_v0_1_server(server) + self.assertNotIn("name", server["packages"][0]) + self.assertNotIn("runtime_hint", server["packages"][0]) + + def test_normalize_v0_1_server_handles_missing_packages(self): + """Server with no packages list passes through unchanged.""" + server = {"name": "io.github.foo/bar", "version": "1.0.0"} + normalized = SimpleRegistryClient._normalize_v0_1_server(server) + self.assertEqual(normalized["name"], "io.github.foo/bar") + self.assertNotIn("packages", normalized) + + def test_normalize_v0_1_server_returns_non_dict_unchanged(self): + self.assertIsNone(SimpleRegistryClient._normalize_v0_1_server(None)) + if __name__ == "__main__": unittest.main() diff --git a/tests/unit/test_registry_client_http_cache.py b/tests/unit/test_registry_client_http_cache.py index 1970a3614..dcdc7e2a9 100644 --- a/tests/unit/test_registry_client_http_cache.py +++ b/tests/unit/test_registry_client_http_cache.py @@ -33,7 +33,7 @@ class TestRegistryHttpCache: def test_fresh_cache_hit_skips_network(self, isolated_cache): """A second list_servers() call within TTL must not hit the network.""" client = SimpleRegistryClient("https://api.mcp.github.com") - body = b'{"servers": [{"name": "a"}], "metadata": {}}' + body = b'{"servers": [{"server": {"name": "a"}}], "metadata": {}}' resp = _mock_response(body=body, headers={"Cache-Control": "max-age=3600"}) with mock.patch.object(client.session, "get", return_value=resp) as mocked: @@ -45,7 +45,7 @@ def test_fresh_cache_hit_skips_network(self, isolated_cache): def test_etag_revalidation_on_304_reuses_body(self, isolated_cache): """When the cache is expired but server returns 304, the cached body is returned.""" client = SimpleRegistryClient("https://api.mcp.github.com") - body = b'{"servers": [{"name": "etag"}], "metadata": {}}' + body = b'{"servers": [{"server": {"name": "etag"}}], "metadata": {}}' first = _mock_response( body=body,