"""End-to-end tests for MCPB bundle deployment."""
import json
import shutil
import subprocess
import tempfile
import time
from pathlib import Path
import pytest
import requests
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
from pytest_httpserver import HTTPServer
from testcontainers.core.container import DockerContainer
from testcontainers.core.waiting_utils import wait_for_logs
from .conftest import (
BASE_IMAGE,
BUNDLE_NAME,
CONTAINER_PORT,
PROJECT_ROOT,
PYTHON_VERSION,
)
# Map Docker arch to uv --python-platform values
_DOCKER_ARCH_TO_UV_PLATFORM = {
"x86_64": "x86_64-unknown-linux-gnu",
"aarch64": "aarch64-unknown-linux-gnu",
}
def _detect_docker_platform() -> str:
"""Detect the Docker daemon's architecture and return a uv platform string."""
result = subprocess.run(
["docker", "info", "--format", "{{.Architecture}}"],
capture_output=True,
text=True,
)
arch = result.stdout.strip()
platform = _DOCKER_ARCH_TO_UV_PLATFORM.get(arch)
if not platform:
raise RuntimeError(
f"Unsupported Docker architecture: {arch}. "
f"Supported: {list(_DOCKER_ARCH_TO_UV_PLATFORM.keys())}"
)
return platform
def build_bundle(output_dir: Path) -> Path:
"""Build MCPB bundle with Linux-compatible deps.
Mirrors the CI pipeline: vendor deps for Linux, then pack.
Uses uv's --python-platform flag for cross-platform dep resolution
so compiled wheels (pydantic_core, etc.) match the Docker container's
OS and architecture, not the host machine's.
"""
# Copy project to a temp build dir so we don't pollute the source tree
build_dir = output_dir / "build"
shutil.copytree(
PROJECT_ROOT,
build_dir,
ignore=shutil.ignore_patterns(".venv", ".git", "*.pyc", "__pycache__", "e2e"),
)
# Vendor deps for the Docker container's platform
deps_dir = build_dir / "deps"
if deps_dir.exists():
shutil.rmtree(deps_dir)
docker_platform = _detect_docker_platform()
result = subprocess.run(
[
"uv",
"pip",
"install",
"--target",
str(deps_dir),
"--python-platform",
docker_platform,
"--python-version",
PYTHON_VERSION,
".",
],
capture_output=True,
text=True,
cwd=build_dir,
)
if result.returncode != 0:
raise RuntimeError(f"Dep vendoring failed: {result.stderr}")
# Read version from manifest for the bundle filename
manifest = json.loads((build_dir / "manifest.json").read_text())
version = manifest["version"]
bundle_path = output_dir / f"{BUNDLE_NAME}-v{version}.mcpb"
result = subprocess.run(
["mcpb", "pack", str(build_dir), str(bundle_path)],
capture_output=True,
text=True,
)
if result.returncode != 0:
raise RuntimeError(f"Bundle build failed: {result.stderr}")
if not bundle_path.exists():
raise RuntimeError(f"Bundle not found at {bundle_path}")
return bundle_path
@pytest.fixture(scope="module")
def bundle_path():
"""Build the MCPB bundle once for all tests."""
with tempfile.TemporaryDirectory() as tmpdir:
path = build_bundle(Path(tmpdir))
# Read bundle content to memory so it survives tmpdir cleanup
content = path.read_bytes()
yield path.name, content
@pytest.fixture(scope="module")
def bundle_server(bundle_path):
"""Serve the bundle over HTTP."""
bundle_name, bundle_content = bundle_path
# Start HTTP server on a random port
server = HTTPServer(host="0.0.0.0", port=0)
server.expect_request(f"/{bundle_name}").respond_with_data(
bundle_content,
content_type="application/octet-stream",
)
server.start()
yield server
server.stop()
@pytest.fixture(scope="module")
def mcpb_container(bundle_server, bundle_path):
"""Run the MCPB container."""
bundle_name, _ = bundle_path
bundle_url = f"http://host.docker.internal:{bundle_server.port}/{bundle_name}"
container = (
DockerContainer(BASE_IMAGE)
.with_env("BUNDLE_URL", bundle_url)
.with_bind_ports(CONTAINER_PORT, None) # Random host port
.with_kwargs(extra_hosts={"host.docker.internal": "host-gateway"})
)
container.start()
try:
# Wait for server to be ready (wait for uvicorn startup message)
wait_for_logs(container, "Uvicorn running on", timeout=60)
# Give uvicorn a moment to fully initialize
time.sleep(1)
# Get the mapped port
host_port = container.get_exposed_port(CONTAINER_PORT)
base_url = f"http://localhost:{host_port}"
# Wait for health endpoint
for _ in range(30):
try:
resp = requests.get(f"{base_url}/health", timeout=2)
if resp.status_code == 200:
break
except requests.RequestException:
time.sleep(0.5)
else:
logs = container.get_logs()
raise RuntimeError(f"Container not healthy. Logs: {logs}")
yield base_url
finally:
container.stop()
def test_health_endpoint(mcpb_container):
"""Test that the health endpoint returns successfully."""
base_url = mcpb_container
response = requests.get(f"{base_url}/health", timeout=5)
assert response.status_code == 200
data = response.json()
assert data["status"] == "healthy"
assert data["service"] == "mcp-echo"
@pytest.mark.asyncio
async def test_mcp_tools_list(mcpb_container):
"""Test that the MCP tools/list endpoint returns all expected tools."""
base_url = mcpb_container
async with streamablehttp_client(f"{base_url}/mcp") as (read, write, _):
async with ClientSession(read, write) as session:
await session.initialize()
tools_response = await session.list_tools()
tools = tools_response.tools
assert tools, "No tools returned from MCP server"
tool_names = {tool.name for tool in tools}
expected_tools = {"echo_message", "echo_with_delay", "echo_json"}
assert expected_tools.issubset(tool_names), (
f"Missing tools: {expected_tools - tool_names}"
)
for tool in tools:
assert tool.name
assert tool.description
assert tool.inputSchema
@pytest.mark.asyncio
async def test_mcp_tool_call(mcpb_container):
"""Test calling an MCP tool."""
base_url = mcpb_container
async with streamablehttp_client(f"{base_url}/mcp") as (read, write, _):
async with ClientSession(read, write) as session:
await session.initialize()
result = await session.call_tool(
"echo_message",
{"message": "Hello from MCPB!", "uppercase": True},
)
assert result.content
assert len(result.content) > 0
# Verify the response contains expected data
text_content = result.content[0].text
assert "HELLO FROM MCPB!" in text_content
