76 lines
2.4 KiB
Python
76 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from services.runtime.connector_stitch import (
|
|
PUBLIC_TO_UPSTREAM,
|
|
UPSTREAM_TO_PUBLIC,
|
|
rewrite_request,
|
|
rewrite_response,
|
|
upstream_headers,
|
|
validate_endpoint,
|
|
)
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def test_stitch_surface_is_curated_and_namespaced() -> None:
|
|
assert len(UPSTREAM_TO_PUBLIC) == len(PUBLIC_TO_UPSTREAM) == 8
|
|
assert all(name.startswith("stitch.") for name in PUBLIC_TO_UPSTREAM)
|
|
assert PUBLIC_TO_UPSTREAM["stitch.screen.generate"] == "generate_screen_from_text"
|
|
|
|
|
|
def test_stitch_rewrites_registered_calls_and_denies_unknown_tools() -> None:
|
|
call = {"method": "tools/call", "params": {"name": "stitch.project.list", "arguments": {}}}
|
|
assert rewrite_request(call)["params"]["name"] == "list_projects"
|
|
with pytest.raises(ValueError, match="not registered"):
|
|
rewrite_request({"method": "tools/call", "params": {"name": "delete_project"}})
|
|
|
|
|
|
def test_stitch_discovery_drops_unreviewed_upstream_tools() -> None:
|
|
response = {"result": {"tools": [{"name": "get_screen"}, {"name": "delete_project"}]}}
|
|
assert rewrite_response(response, "tools/list")["result"]["tools"] == [
|
|
{"name": "stitch.screen.get"}
|
|
]
|
|
|
|
|
|
def test_stitch_screen_get_manifest_matches_live_google_mcp_contract() -> None:
|
|
manifest = json.loads(
|
|
(ROOT / "catalog/tools/stitch-screen-get.yaml").read_text(encoding="utf-8")
|
|
)
|
|
schema = manifest["spec"]["inputSchema"]
|
|
|
|
assert schema["required"] == ["name"]
|
|
assert set(schema["properties"]) == {"name"}
|
|
|
|
|
|
def test_stitch_api_key_and_only_mcp_headers_reach_google() -> None:
|
|
headers = upstream_headers(
|
|
{"mcp-session-id": "s1", "authorization": "must-not-leak", "cookie": "no"},
|
|
"google-key",
|
|
has_body=True,
|
|
)
|
|
assert headers == {
|
|
"X-Goog-Api-Key": "google-key",
|
|
"Accept": "application/json, text/event-stream",
|
|
"Content-Type": "application/json",
|
|
"mcp-session-id": "s1",
|
|
}
|
|
|
|
|
|
def test_stitch_endpoint_is_pinned_to_google() -> None:
|
|
assert (
|
|
validate_endpoint("https://stitch.googleapis.com/mcp")
|
|
== "https://stitch.googleapis.com/mcp"
|
|
)
|
|
for value in (
|
|
"http://stitch.googleapis.com/mcp",
|
|
"https://evil.example/mcp",
|
|
"https://stitch.googleapis.com/v1",
|
|
):
|
|
with pytest.raises(ValueError):
|
|
validate_endpoint(value)
|