diff --git a/pyproject.toml b/pyproject.toml index 2c3a44a..429710e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -86,6 +86,10 @@ ignore = [ "B008", # function calls in argument defaults (pydantic Field style) ] +[tool.ruff.lint.isort] +# tests/mcp is a directory, not the SDK: keep "mcp" classified as third party. +known-third-party = ["mcp"] + [tool.ruff.lint.per-file-ignores] "__init__.py" = ["F401"] "tests/integration/harness.py" = [ @@ -129,6 +133,7 @@ pythonpath = ["src"] asyncio_mode = "auto" markers = [ "integration: needs a live Gitea in Docker (opt-in via AEGIS_INTEGRATION=1)", + "mcp_smoke: MCP client smoke tests over real transports (opt-in via AEGIS_INTEGRATION=1)", ] [tool.coverage.run] diff --git a/tests/mcp/__init__.py b/tests/mcp/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/mcp/conftest.py b/tests/mcp/conftest.py new file mode 100644 index 0000000..6e59e96 --- /dev/null +++ b/tests/mcp/conftest.py @@ -0,0 +1,76 @@ +"""Opt-in fixtures for the MCP smoke tests. + +Tests in this directory are marked ``mcp_smoke`` and skipped unless +``AEGIS_INTEGRATION=1``. They reuse the live-Gitea harness from +``tests/integration`` (one disposable Gitea per session, admin token with the +``all`` scope) and drive the server through a real MCP client over stdio and +streamable HTTP, from source and (when ``AEGIS_MCP_IMAGE`` is set) from the +built image. +""" + +from __future__ import annotations + +import os +import secrets +from pathlib import Path + +import pytest + +from tests.integration.conftest import ENABLE_ENV, gitea, integration_enabled +from tests.integration.harness import GITEA_PORT, GiteaInstance + +__all__ = ["ENABLE_ENV", "gitea"] + +_HERE = Path(__file__).parent.resolve() +IMAGE_ENV = "AEGIS_MCP_IMAGE" + + +def pytest_configure(config: pytest.Config) -> None: + """Register the ``mcp_smoke`` marker (idempotent with pyproject).""" + config.addinivalue_line( + "markers", "mcp_smoke: MCP client smoke tests, opt-in via AEGIS_INTEGRATION=1" + ) + + +def _is_mcp_smoke(item: pytest.Item) -> bool: + try: + return _HERE in Path(str(item.path)).resolve().parents + except (OSError, ValueError): # pragma: no cover - defensive + return False + + +def pytest_collection_modifyitems(config: pytest.Config, items: list[pytest.Item]) -> None: + """Mark and (unless opted in) skip every test under tests/mcp.""" + selected = [item for item in items if _is_mcp_smoke(item)] + skip = pytest.mark.skip(reason=f"MCP smoke tests need {ENABLE_ENV}=1 and Docker") + for item in selected: + item.add_marker(pytest.mark.mcp_smoke) + if not integration_enabled(): + item.add_marker(skip) + if selected and len(selected) == len(items): + # A smoke-only run starts the server in subprocesses, so the in-process + # coverage gate does not apply. + for namespace in (config.option, config.known_args_namespace): + if getattr(namespace, "cov_fail_under", None) is not None: + namespace.cov_fail_under = 0 + + +@pytest.fixture(scope="session") +def gitea_internal_url(gitea: GiteaInstance) -> str: + """URL of the Gitea container for other containers on the harness network.""" + return f"http://{gitea.container}:{GITEA_PORT}" + + +@pytest.fixture(scope="session") +def mcp_image() -> str: + """The built image under test; the test is skipped when ``AEGIS_MCP_IMAGE`` is unset.""" + image = os.environ.get(IMAGE_ENV, "").strip() + if not image: + pytest.skip(f"{IMAGE_ENV} is not set; image smoke tests need a built image") + return image + + +@pytest.fixture +def bearer_token() -> str: + """A random 48 character MCP bearer token.""" + return secrets.token_urlsafe(36) diff --git a/tests/mcp/helpers.py b/tests/mcp/helpers.py new file mode 100644 index 0000000..dc06615 --- /dev/null +++ b/tests/mcp/helpers.py @@ -0,0 +1,425 @@ +"""Shared helpers for the MCP smoke tests: client flows, subprocess and docker plumbing. + +Secrets (Gitea token, MCP bearer token) are always passed through environment +variables, never through argv, and every captured stderr/stdout is checked for +them by the tests. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import json +import os +import secrets +import socket +import subprocess +import sys +import time +from collections.abc import Iterator, Mapping, Sequence +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +import httpx +import mcp.types as types +from mcp import Client + +from aegis_gitea_mcp.catalog import load_catalog +from tests.integration.harness import NAME_PREFIX, GiteaInstance + +# Raw-protocol version used for the hand-driven stdio session (handshake era). +RAW_PROTOCOL_VERSION = "2025-11-25" +SOURCE_SERVE_ARGS = [sys.executable, "-m", "aegis_gitea_mcp.cli", "serve"] +HEALTH_TIMEOUT_SECONDS = 60.0 +RAW_READ_TIMEOUT_SECONDS = 60.0 +# Environment variables a child process needs to find its interpreter and docker. +_PASSTHROUGH_ENV = ( + "PATH", + "HOME", + "LANG", + "LC_ALL", + "TMPDIR", + "DOCKER_HOST", + "DOCKER_CONFIG", + "DOCKER_CONTEXT", + "XDG_RUNTIME_DIR", +) + + +def expected_tool_count() -> int: + """Number of tools a full server exposes: every catalog operation plus the raw tool.""" + return len(load_catalog().operations) + 1 + + +def unique_container_name() -> str: + """Return a unique ``aegis-it-mcp-`` container name.""" + return f"{NAME_PREFIX}mcp-{secrets.token_hex(4)}" + + +def free_loopback_port() -> int: + """Return a free TCP port on 127.0.0.1 (same approach as the Gitea harness).""" + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.bind(("127.0.0.1", 0)) + return int(sock.getsockname()[1]) + + +def child_env(extra: Mapping[str, str]) -> dict[str, str]: + """Minimal environment for a child process: passthrough variables plus ``extra``.""" + env = {name: os.environ[name] for name in _PASSTHROUGH_ENV if name in os.environ} + env.update(extra) + return env + + +def server_env( + gitea: GiteaInstance, gitea_url: str, data_dir: Path, **extra: str +) -> dict[str, str]: + """Server settings for a subprocess (``GITEA_TOKEN`` included, so keep it out of argv).""" + return child_env( + { + "GITEA_URL": gitea_url, + "GITEA_TOKEN": gitea.admin_token, + "DATA_DIR": str(data_dir), + "LOG_LEVEL": "DEBUG", + **extra, + } + ) + + +# --------------------------------------------------------------------------- +# Client flows +# --------------------------------------------------------------------------- + + +def result_text(result: types.CallToolResult) -> str: + """Return the text of the first content block of a tool result.""" + assert result.content, "tool result has no content" + block = result.content[0] + assert isinstance(block, types.TextContent) + return block.text + + +def result_payload(result: types.CallToolResult) -> dict[str, Any]: + """Parse the JSON envelope of a tool result.""" + payload = json.loads(result_text(result)) + assert isinstance(payload, dict) + return payload + + +async def exercise_server( + client: Client, gitea: GiteaInstance, secrets_to_hide: Sequence[str] +) -> None: + """List tools, run a read call and a write call (create then delete a repo). + + Asserts the tool count, the identity in the envelope, that the repo really + existed on Gitea after creation and was gone after deletion, and that no + secret appears in any tool result. + """ + listing = await client.list_tools() + names = {tool.name for tool in listing.tools} + assert len(listing.tools) == expected_tool_count() + assert {"gitea_user_get_current", "gitea_repo_delete", "gitea_api_request"} <= names + + texts: list[str] = [] + + current = await client.call_tool("gitea_user_get_current", {}) + texts.append(result_text(current)) + assert not current.is_error + envelope = result_payload(current) + assert envelope["status"] == 200 + assert envelope["identity"]["login"] == gitea.admin_login + assert envelope["identity"]["mode"] == "admin-token" + assert envelope["data"]["login"] == gitea.admin_login + + repo = gitea.unique("mcp-smoke") + created = False + try: + create = await client.call_tool( + "gitea_user_current_create_repo", + {"body": {"name": repo, "auto_init": False, "private": True}}, + ) + texts.append(result_text(create)) + assert not create.is_error, result_text(create) + created = True + assert result_payload(create)["status"] == 201 + assert gitea.api("GET", f"/repos/{gitea.admin_login}/{repo}").status_code == 200 + + delete = await client.call_tool( + "gitea_repo_delete", {"owner": gitea.admin_login, "repo": repo} + ) + texts.append(result_text(delete)) + assert not delete.is_error, result_text(delete) + created = False + assert gitea.api("GET", f"/repos/{gitea.admin_login}/{repo}").status_code == 404 + finally: + if created: + gitea.api("DELETE", f"/repos/{gitea.admin_login}/{repo}") + + for secret in secrets_to_hide: + assert all(secret not in text for text in texts), "a secret leaked into a tool result" + + +def assert_no_secrets(text: str, secrets_to_hide: Sequence[str], where: str) -> None: + """Fail when any secret appears in ``text`` (the message never prints the secret).""" + for index, secret in enumerate(secrets_to_hide): + assert secret not in text, f"secret #{index} found in {where}" + + +# --------------------------------------------------------------------------- +# Hand-driven stdio session (checks that stdout carries JSON-RPC only) +# --------------------------------------------------------------------------- + + +@dataclass +class RawSession: + """Everything a hand-driven stdio run produced.""" + + stdout_lines: list[str] + stderr: str + returncode: int | None + + +async def _read_response(stdout: asyncio.StreamReader, request_id: int, lines: list[str]) -> None: + """Read stdout lines until the response for ``request_id`` arrived (lines are recorded).""" + while True: + raw = await asyncio.wait_for(stdout.readline(), timeout=RAW_READ_TIMEOUT_SECONDS) + if not raw: + raise AssertionError("server closed stdout before answering") + line = raw.decode("utf-8").rstrip("\r\n") + lines.append(line) + message = json.loads(line) + if ( + isinstance(message, dict) + and message.get("id") == request_id + and "method" not in message + ): + assert "result" in message, f"request {request_id} failed" + return + + +async def run_raw_stdio( + argv: Sequence[str], env: Mapping[str, str], cwd: Path | None = None +) -> RawSession: + """Start ``argv`` and run initialize + tools/list by hand, then close stdin. + + Returns every stdout line the process wrote (so the caller can assert that + each one is a JSON-RPC message), its stderr and the exit code. + """ + proc = await asyncio.create_subprocess_exec( + *argv, + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + env=dict(env), + cwd=cwd, + limit=64 * 1024 * 1024, + ) + assert proc.stdin is not None + assert proc.stdout is not None + assert proc.stderr is not None + lines: list[str] = [] + stderr_task = asyncio.create_task(proc.stderr.read()) + try: + + async def send(message: dict[str, Any]) -> None: + proc.stdin.write((json.dumps(message) + "\n").encode("utf-8")) # type: ignore[union-attr] + await proc.stdin.drain() # type: ignore[union-attr] + + await send( + { + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": { + "protocolVersion": RAW_PROTOCOL_VERSION, + "capabilities": {}, + "clientInfo": {"name": "aegis-smoke", "version": "0"}, + }, + } + ) + await _read_response(proc.stdout, 1, lines) + await send({"jsonrpc": "2.0", "method": "notifications/initialized"}) + await send({"jsonrpc": "2.0", "id": 2, "method": "tools/list"}) + await _read_response(proc.stdout, 2, lines) + proc.stdin.close() + rest = await asyncio.wait_for(proc.stdout.read(), timeout=30.0) + lines.extend(rest.decode("utf-8").splitlines()) + await asyncio.wait_for(proc.wait(), timeout=30.0) + finally: + if proc.returncode is None: + with contextlib.suppress(ProcessLookupError): + proc.kill() + await proc.wait() + stderr = (await stderr_task).decode("utf-8", errors="replace") + return RawSession(stdout_lines=lines, stderr=stderr, returncode=proc.returncode) + + +def assert_jsonrpc_only(lines: Sequence[str]) -> None: + """Assert that every non-empty stdout line is a JSON-RPC 2.0 message.""" + assert lines, "no output on stdout" + for line in lines: + if not line.strip(): + continue + message = json.loads(line) + assert isinstance(message, dict) and message.get("jsonrpc") == "2.0" + + +# --------------------------------------------------------------------------- +# HTTP readiness +# --------------------------------------------------------------------------- + + +def wait_for_health( + url: str, *, alive: Any = None, timeout: float = HEALTH_TIMEOUT_SECONDS +) -> httpx.Response: + """Poll ``url`` until it answers 200 and return the response. + + ``alive`` is an optional callable returning False when the server died, so + the wait fails fast instead of running into the timeout. + """ + deadline = time.monotonic() + timeout + last = "no response" + while time.monotonic() < deadline: + if alive is not None and not alive(): + raise AssertionError("the server process exited before /health answered") + try: + response = httpx.get(url, timeout=3.0) + if response.status_code == 200: + return response + last = f"HTTP {response.status_code}" + except httpx.HTTPError as exc: + last = type(exc).__name__ + time.sleep(0.5) + raise AssertionError(f"{url} was not healthy within {timeout:.0f}s ({last})") + + +def assert_rejects_without_valid_token(mcp_url: str, bearer_token: str) -> None: + """A missing and a wrong bearer token both get 401 (and the real token never echoes back).""" + initialize = { + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": { + "protocolVersion": RAW_PROTOCOL_VERSION, + "capabilities": {}, + "clientInfo": {"name": "aegis-smoke", "version": "0"}, + }, + } + base = {"Accept": "application/json, text/event-stream"} + anonymous = httpx.post(mcp_url, json=initialize, headers=base, timeout=10.0) + assert anonymous.status_code == 401 + wrong = httpx.post( + mcp_url, + json=initialize, + headers={**base, "Authorization": f"Bearer {secrets.token_urlsafe(36)}"}, + timeout=10.0, + ) + assert wrong.status_code == 401 + assert bearer_token not in anonymous.text + wrong.text + + +# --------------------------------------------------------------------------- +# Source HTTP server +# --------------------------------------------------------------------------- + + +@dataclass +class HttpServer: + """A source-installed HTTP server subprocess.""" + + process: subprocess.Popen[bytes] + base_url: str + stderr_path: Path + stdout_path: Path + + @property + def mcp_url(self) -> str: + """URL of the MCP endpoint.""" + return f"{self.base_url}/mcp" + + def alive(self) -> bool: + """Return True while the process is running.""" + return self.process.poll() is None + + def stderr_text(self) -> str: + """Return what the server wrote to stderr so far.""" + return self.stderr_path.read_text(encoding="utf-8", errors="replace") + + +@contextlib.contextmanager +def run_source_http_server(env: Mapping[str, str], workdir: Path) -> Iterator[HttpServer]: + """Run ``aegis-gitea-mcp serve --transport http`` on a free loopback port. + + The working directory is an empty temp dir so no ambient ``.env`` is read. + Waits for ``/health`` and always stops the process afterwards. + """ + port = free_loopback_port() + stderr_path = workdir / "server.stderr" + stdout_path = workdir / "server.stdout" + with stderr_path.open("wb") as err, stdout_path.open("wb") as out: + process = subprocess.Popen( # noqa: S603 + [ + *SOURCE_SERVE_ARGS, + "--transport", + "http", + "--host", + "127.0.0.1", + "--port", + str(port), + ], + env=dict(env), + cwd=workdir, + stdin=subprocess.DEVNULL, + stdout=out, + stderr=err, + ) + server = HttpServer( + process=process, + base_url=f"http://127.0.0.1:{port}", + stderr_path=stderr_path, + stdout_path=stdout_path, + ) + try: + wait_for_health(f"{server.base_url}/health", alive=server.alive) + yield server + finally: + process.terminate() + try: + process.wait(timeout=15) + except subprocess.TimeoutExpired: + process.kill() + process.wait(timeout=15) + + +# --------------------------------------------------------------------------- +# Docker +# --------------------------------------------------------------------------- + + +def docker( + *args: str, + env: Mapping[str, str] | None = None, + check: bool = True, + timeout: float = 120.0, +) -> subprocess.CompletedProcess[str]: + """Run ``docker ``; ``env`` adds variables the CLI can forward with ``-e NAME``.""" + full_env = {**os.environ, **(env or {})} + proc = subprocess.run( # noqa: S603 + ["docker", *args], # noqa: S607 + capture_output=True, + text=True, + timeout=timeout, + check=False, + env=full_env, + ) + if check and proc.returncode != 0: + # Never echo argv: it may carry non-secret config, and stderr is enough to debug. + raise AssertionError( + f"docker {' '.join(args[:2])} failed (exit {proc.returncode}): {proc.stderr.strip()}" + ) + return proc + + +def remove_container(name: str) -> None: + """Force-remove one container by its exact name (best effort, never prunes).""" + docker("rm", "-f", "-v", name, check=False, timeout=60) diff --git a/tests/mcp/test_http.py b/tests/mcp/test_http.py new file mode 100644 index 0000000..bb25913 --- /dev/null +++ b/tests/mcp/test_http.py @@ -0,0 +1,46 @@ +"""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") diff --git a/tests/mcp/test_image.py b/tests/mcp/test_image.py new file mode 100644 index 0000000..051d898 --- /dev/null +++ b/tests/mcp/test_image.py @@ -0,0 +1,212 @@ +"""MCP smoke tests against the built Docker image (``AEGIS_MCP_IMAGE``). + +The MCP container joins the harness's per-run network and reaches Gitea by its +container name. Secrets go into the container with ``-e NAME`` (the value comes +from the docker CLI's environment, never from argv). Every container started +here has a unique ``aegis-it-mcp-`` name and is removed in a ``finally``. +""" + +from __future__ import annotations + +import contextlib +import os +from collections.abc import Iterator +from pathlib import Path + +import httpx2 +import pytest +from mcp import Client, StdioServerParameters +from mcp.client.stdio import stdio_client + +from tests.integration.harness import NETWORK_MODE_ENV, GiteaInstance +from tests.mcp.helpers import ( + assert_jsonrpc_only, + assert_no_secrets, + assert_rejects_without_valid_token, + child_env, + docker, + exercise_server, + expected_tool_count, + free_loopback_port, + remove_container, + run_raw_stdio, + unique_container_name, + wait_for_health, +) + +CONTAINER_HTTP_PORT = 8080 +HTTP_TIMEOUT_SECONDS = 60.0 + + +@pytest.fixture(scope="module", autouse=True) +def _image_present(mcp_image: str) -> None: + """Fail (not skip) when the configured image does not exist locally.""" + proc = docker("image", "inspect", mcp_image, check=False, timeout=60) + assert proc.returncode == 0, f"AEGIS_MCP_IMAGE={mcp_image!r} is not available locally" + + +def _stdio_argv(name: str, network: str, gitea_url: str, image: str) -> list[str]: + return [ + "run", + "--rm", + "-i", + "--name", + name, + "--network", + network, + "--label", + "aegis-it=1", + "-e", + f"GITEA_URL={gitea_url}", + "-e", + "GITEA_TOKEN", + image, + "serve", + "--transport", + "stdio", + ] + + +async def test_image_stdio_lists_reads_and_writes( + gitea: GiteaInstance, gitea_internal_url: str, mcp_image: str, tmp_path: Path +) -> None: + """The image over stdio (docker run -i): list, read, create and delete a repo.""" + name = unique_container_name() + argv = _stdio_argv(name, gitea.network, gitea_internal_url, mcp_image) + params = StdioServerParameters( + command="docker", args=argv, env=child_env({"GITEA_TOKEN": gitea.admin_token}) + ) + stderr_path = tmp_path / "container.stderr" + try: + with stderr_path.open("w", encoding="utf-8") as errlog: + async with Client(stdio_client(params, errlog=errlog), read_timeout_seconds=60.0) as c: + await exercise_server(c, gitea, [gitea.admin_token]) + finally: + remove_container(name) + assert_no_secrets( + stderr_path.read_text(encoding="utf-8", errors="replace"), + [gitea.admin_token], + "the container stderr", + ) + + +async def test_image_stdio_stdout_is_json_rpc_only( + gitea: GiteaInstance, gitea_internal_url: str, mcp_image: str +) -> None: + """Nothing but JSON-RPC reaches the image's stdout; stderr carries no secret.""" + name = unique_container_name() + argv = ["docker", *_stdio_argv(name, gitea.network, gitea_internal_url, mcp_image)] + try: + session = await run_raw_stdio(argv, child_env({"GITEA_TOKEN": gitea.admin_token})) + finally: + remove_container(name) + assert_jsonrpc_only(session.stdout_lines) + assert session.returncode == 0 + assert_no_secrets(session.stderr, [gitea.admin_token], "the container stderr") + + +@contextlib.contextmanager +def _image_http_server( + gitea: GiteaInstance, gitea_internal_url: str, image: str, bearer_token: str +) -> Iterator[tuple[str, str]]: + """Run the image in HTTP mode; yield ``(base_url, container_name)``. + + Local mode publishes the port on 127.0.0.1 only; container mode (the job + container shares the per-run network) reaches the MCP container by name. + """ + name = unique_container_name() + container_mode = os.environ.get(NETWORK_MODE_ENV, "").lower() == "container" + if container_mode: + base_url = f"http://{name}:{CONTAINER_HTTP_PORT}" + allowed_host = f"{name}:{CONTAINER_HTTP_PORT}" + publish: list[str] = [] + else: + port = free_loopback_port() + base_url = f"http://127.0.0.1:{port}" + allowed_host = f"127.0.0.1:{port}" + publish = ["-p", f"127.0.0.1:{port}:{CONTAINER_HTTP_PORT}"] + try: + docker( + "run", + "-d", + "--name", + name, + "--network", + gitea.network, + "--label", + "aegis-it=1", + *publish, + "-e", + f"GITEA_URL={gitea_internal_url}", + "-e", + f"ALLOWED_HOSTS={allowed_host}", + "-e", + "GITEA_TOKEN", + "-e", + "MCP_BEARER_TOKEN", + image, + env={"GITEA_TOKEN": gitea.admin_token, "MCP_BEARER_TOKEN": bearer_token}, + ) + + def alive() -> bool: + state = docker("inspect", "--format", "{{.State.Running}}", name, check=False) + return state.stdout.strip() == "true" + + wait_for_health(f"{base_url}/health", alive=alive) + yield base_url, name + finally: + remove_container(name) + + +async def test_image_http_lists_reads_and_writes( + gitea: GiteaInstance, gitea_internal_url: str, mcp_image: str, bearer_token: str +) -> None: + """The image over streamable HTTP: list, read, write, 401 negatives, no secret in the logs.""" + from mcp.client.streamable_http import streamable_http_client + + hidden = [gitea.admin_token, bearer_token] + with _image_http_server(gitea, gitea_internal_url, mcp_image, bearer_token) as (base, name): + health = httpx2.get(f"{base}/health", timeout=10.0).json() + assert health["status"] == "ok" + assert health["tools"] == expected_tool_count() + + async with httpx2.AsyncClient( + headers={"Authorization": f"Bearer {bearer_token}"}, timeout=HTTP_TIMEOUT_SECONDS + ) as http_client: + transport = streamable_http_client(f"{base}/mcp", 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(f"{base}/mcp", bearer_token) + logs = docker("logs", name, check=False, timeout=60) + assert_no_secrets(logs.stdout + logs.stderr, hidden, "the container logs") + + +def test_image_refuses_open_bind_without_auth( + gitea: GiteaInstance, gitea_internal_url: str, mcp_image: str +) -> None: + """The image binds 0.0.0.0 by default and must exit 2 without MCP_BEARER_TOKEN.""" + name = unique_container_name() + try: + proc = docker( + "run", + "--rm", + "--name", + name, + "--label", + "aegis-it=1", + "-e", + f"GITEA_URL={gitea_internal_url}", + "-e", + "GITEA_TOKEN", + mcp_image, + "serve", + env={"GITEA_TOKEN": gitea.admin_token}, + check=False, + timeout=90, + ) + finally: + remove_container(name) + assert proc.returncode == 2 + assert "MCP_BEARER_TOKEN" in proc.stderr + assert_no_secrets(proc.stdout + proc.stderr, [gitea.admin_token], "the container output") diff --git a/tests/mcp/test_stdio.py b/tests/mcp/test_stdio.py new file mode 100644 index 0000000..f0688c0 --- /dev/null +++ b/tests/mcp/test_stdio.py @@ -0,0 +1,53 @@ +"""MCP smoke tests: the stdio transport from source, against a live Gitea.""" + +from __future__ import annotations + +from pathlib import Path + +from mcp import Client, StdioServerParameters +from mcp.client.stdio import stdio_client + +from tests.integration.harness import GiteaInstance +from tests.mcp.helpers import ( + SOURCE_SERVE_ARGS, + assert_jsonrpc_only, + assert_no_secrets, + exercise_server, + server_env, +) + + +def _stdio_params(gitea: GiteaInstance, data_dir: Path) -> StdioServerParameters: + return StdioServerParameters( + command=SOURCE_SERVE_ARGS[0], + args=[*SOURCE_SERVE_ARGS[1:], "--transport", "stdio"], + # The working directory is the empty temp dir so no ambient .env is read. + env=server_env(gitea, gitea.base_url, data_dir), + cwd=data_dir, + ) + + +async def test_stdio_lists_reads_and_writes(gitea: GiteaInstance, tmp_path: Path) -> None: + """A real MCP client over stdio: initialize, list 538 tools, read, create and delete a repo.""" + stderr_path = tmp_path / "server.stderr" + with stderr_path.open("w", encoding="utf-8") as errlog: + transport = stdio_client(_stdio_params(gitea, tmp_path), errlog=errlog) + async with Client(transport, read_timeout_seconds=60.0) as client: + assert client.server_info is not None + assert client.server_info.name == "aegis-gitea-mcp" + await exercise_server(client, gitea, [gitea.admin_token]) + + stderr = stderr_path.read_text(encoding="utf-8", errors="replace") + assert_no_secrets(stderr, [gitea.admin_token], "the server stderr") + + +async def test_stdio_stdout_is_json_rpc_only(gitea: GiteaInstance, tmp_path: Path) -> None: + """Every stdout line of the stdio server is JSON-RPC; logs go to stderr without secrets.""" + from tests.mcp.helpers import run_raw_stdio + + params = _stdio_params(gitea, tmp_path) + session = await run_raw_stdio([params.command, *params.args], params.env or {}, tmp_path) + assert_jsonrpc_only(session.stdout_lines) + assert session.returncode == 0 + assert_no_secrets("\n".join(session.stdout_lines), [gitea.admin_token], "the server stdout") + assert_no_secrets(session.stderr, [gitea.admin_token], "the server stderr")