75 lines
3.4 KiB
Python
75 lines
3.4 KiB
Python
# #region Tooling.McpProxySmoke [C:3] [TYPE Module] [SEMANTICS docker,mcp,nginx,verification]
|
|
# @BRIEF Isolated real nginx smoke for HTTP/TLS MCP routing, authority and protocol headers.
|
|
# @POST Containers and network created by this invocation are removed, including on assertion failure.
|
|
# @REJECTED Reusing the development compose project could attach tests to the operator database.
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import subprocess
|
|
import time
|
|
import urllib.request
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
|
|
# #region Tooling.McpProxySmoke.Command [C:2] [TYPE Function]
|
|
def docker(*args: str) -> str:
|
|
return subprocess.check_output(["docker", *args], text=True).strip()
|
|
# #endregion Tooling.McpProxySmoke.Command
|
|
|
|
|
|
# #region Tooling.McpProxySmoke.Run [C:3] [TYPE Function]
|
|
def main() -> None:
|
|
root = Path(__file__).resolve().parents[1]
|
|
prefix = "ss-mcp-proxy-smoke-" + uuid.uuid4().hex[:10]
|
|
network, backend, frontend = prefix, prefix + "-backend", prefix + "-frontend"
|
|
try:
|
|
docker("network", "create", network)
|
|
docker(
|
|
"run", "-d", "--name", backend, "--network", network, "--network-alias", "backend",
|
|
"-v", f"{root / 'scripts/mcp_proxy_fixture.py'}:/fixture.py:ro",
|
|
"python:3.13-alpine", "python", "/fixture.py",
|
|
)
|
|
docker(
|
|
"run", "-d", "--name", frontend, "--network", network,
|
|
"-p", "127.0.0.1::80",
|
|
"-v", f"{root / 'docker/nginx.conf'}:/etc/nginx/conf.d/default.conf:ro",
|
|
"nginx:1.27-alpine",
|
|
)
|
|
docker("exec", frontend, "nginx", "-t")
|
|
port = docker("port", frontend, "80/tcp").rsplit(":", 1)[1]
|
|
origin = f"http://127.0.0.1:{port}"
|
|
for _ in range(50):
|
|
try:
|
|
urllib.request.urlopen(origin + "/.well-known/oauth-authorization-server", timeout=2)
|
|
break
|
|
except OSError:
|
|
time.sleep(0.2)
|
|
for path, method in [
|
|
("/.well-known/oauth-authorization-server", "GET"),
|
|
("/.well-known/oauth-protected-resource/mcp", "GET"),
|
|
("/oauth/token", "POST"), ("/oauth/register", "POST"),
|
|
("/oauth/register/fixture", "DELETE"), ("/mcp", "POST"), ("/mcp/", "GET"),
|
|
]:
|
|
request = urllib.request.Request(
|
|
origin + path, method=method,
|
|
data=b'{"jsonrpc":"2.0"}' if method == "POST" else None,
|
|
headers={"Authorization": "Bearer smoke-fixture", "Mcp-Session-Id": "fixture-session"},
|
|
)
|
|
with urllib.request.urlopen(request, timeout=5) as response:
|
|
payload = json.load(response)
|
|
assert response.headers["Mcp-Session-Id"] == "fixture-session"
|
|
assert payload == {"path": path, "method": method, "host": f"127.0.0.1:{port}",
|
|
"scheme": "http", "authorization": True, "session": "fixture-session"}
|
|
print("PASS: nginx routes discovery, OAuth and MCP with external port, bearer and session headers")
|
|
finally:
|
|
for name in (frontend, backend):
|
|
subprocess.run(["docker", "rm", "-f", name], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
subprocess.run(["docker", "network", "rm", network], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
# #endregion Tooling.McpProxySmoke.Run
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
# #endregion Tooling.McpProxySmoke
|