Latte
5e144772b2
ci / gitleaks (push) Successful in 9s
ci / docs-check (push) Failing after 30s
ci / lint (push) Successful in 41s
ci / test-integration (push) Successful in 46s
ci / security (push) Failing after 28s
ci / test-mcp (push) Successful in 1m16s
ci / test-unit (push) Successful in 1m52s
ci / build-and-push (push) Successful in 17s
47 lines
1.7 KiB
Python
47 lines
1.7 KiB
Python
"""MCP smoke tests: the streamable HTTP transport from source, against a live Gitea."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import httpx2
|
|
from mcp import Client
|
|
from mcp.client.streamable_http import streamable_http_client
|
|
|
|
from tests.integration.harness import GiteaInstance
|
|
from tests.mcp.helpers import (
|
|
assert_no_secrets,
|
|
assert_rejects_without_valid_token,
|
|
exercise_server,
|
|
expected_tool_count,
|
|
run_source_http_server,
|
|
server_env,
|
|
)
|
|
|
|
|
|
async def test_http_lists_reads_and_writes(
|
|
gitea: GiteaInstance, tmp_path: Path, bearer_token: str
|
|
) -> None:
|
|
"""A real MCP client over streamable HTTP with a bearer token, plus the 401 negatives."""
|
|
env = server_env(gitea, gitea.base_url, tmp_path, MCP_BEARER_TOKEN=bearer_token)
|
|
hidden = [gitea.admin_token, bearer_token]
|
|
with run_source_http_server(env, tmp_path) as server:
|
|
health = server.base_url + "/health"
|
|
body = httpx2.get(health, timeout=10.0).json()
|
|
assert body["status"] == "ok"
|
|
assert body["tools"] == expected_tool_count()
|
|
|
|
async with httpx2.AsyncClient(
|
|
headers={"Authorization": f"Bearer {bearer_token}"}, timeout=60.0
|
|
) as http_client:
|
|
transport = streamable_http_client(server.mcp_url, http_client=http_client)
|
|
async with Client(transport, read_timeout_seconds=60.0) as client:
|
|
await exercise_server(client, gitea, hidden)
|
|
|
|
assert_rejects_without_valid_token(server.mcp_url, bearer_token)
|
|
assert server.alive()
|
|
|
|
assert_no_secrets(server.stderr_text(), hidden, "the server stderr")
|
|
stdout = server.stdout_path.read_text(encoding="utf-8", errors="replace")
|
|
assert_no_secrets(stdout, hidden, "the server stdout")
|