"""Protocol smoke tests for MCP server.
These tests spawn the MCP server as a subprocess and test it like a real client would,
verifying server startup, lifecycle, and basic process management.
This catches framing/stdio quirks, env var wiring, and lifecycle issues that
in-process tests might miss.
"""
import subprocess
import sys
import time
import pytest
# Mark all tests in this module as protocol tests
pytestmark = [pytest.mark.protocol, pytest.mark.slow]
class TestMCPProtocolSmoke:
"""Black-box protocol smoke tests using subprocess."""
@pytest.fixture
def server_process(self):
"""Start MCP server as subprocess and yield for testing."""
# Start the server in stdio mode
proc = subprocess.Popen(
[sys.executable, "-m", "aerospace_mcp.fastmcp_server"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1,
)
# Give server time to initialize (minimal wait for CI efficiency)
time.sleep(0.3)
yield proc
# Cleanup
proc.terminate()
try:
proc.wait(timeout=5)
except subprocess.TimeoutExpired:
proc.kill()
def test_server_starts(self, server_process):
"""Test that server process starts and is running."""
proc = server_process
assert proc.poll() is None, "Server should be running"
@pytest.mark.skip(reason="MCP protocol requires proper framing - needs MCP client")
def test_list_tools_returns_expected_tools(self, server_process):
"""Test that list_tools returns the expected aerospace tools."""
# This would require proper MCP client implementation with
# Content-Length framing per the MCP protocol specification.
# For now, we verify the server is running.
proc = server_process
assert proc.poll() is None
class TestMCPServerStartup:
"""Test server startup and initialization."""
def test_server_module_is_importable(self):
"""Test that the server module can be imported."""
from aerospace_mcp import fastmcp_server
assert hasattr(fastmcp_server, "mcp")
assert hasattr(fastmcp_server, "run")
def test_server_has_expected_tools_registered(self):
"""Test that all expected tools are registered."""
from aerospace_mcp.fastmcp_server import mcp
# Get tool names from the server
# FastMCP stores tools differently, we need to check the registry
assert mcp is not None
assert mcp.name == "aerospace-mcp"
def test_server_entry_point_exists(self):
"""Test that the console script entry point works."""
result = subprocess.run(
[sys.executable, "-m", "aerospace_mcp.fastmcp_server", "--help"],
capture_output=True,
text=True,
timeout=10,
)
# Server doesn't have --help, but should exit cleanly or run
# We just verify it doesn't crash on import
assert result.returncode in [0, 1, 2] # Various acceptable exit codes
class TestMCPClientIntegration:
"""Integration tests using MCP client library if available."""
@pytest.fixture
def mcp_client(self):
"""Create an MCP client for testing."""
try:
from mcp import Client
return Client
except ImportError:
pytest.skip("MCP client library not available")
@pytest.mark.skip(reason="Requires async MCP client setup")
async def test_client_can_connect_and_list_tools(self, mcp_client):
"""Test that an MCP client can connect and list tools."""
# This would be implemented with proper async MCP client
pass
class TestHTTPAPISmoke:
"""Smoke tests for the HTTP API server."""
@pytest.fixture
def http_server(self):
"""Start HTTP server on ephemeral port."""
import socket
# Find an available port
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(("", 0))
port = s.getsockname()[1]
proc = subprocess.Popen(
[
sys.executable,
"-c",
f"""
import uvicorn
from main import app
uvicorn.run(app, host="127.0.0.1", port={port}, log_level="error")
""",
],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
# Wait for server to start (reduced for CI efficiency)
time.sleep(1)
yield f"http://127.0.0.1:{port}", proc
proc.terminate()
try:
proc.wait(timeout=5)
except subprocess.TimeoutExpired:
proc.kill()
def test_health_endpoint(self, http_server):
"""Test that health endpoint responds."""
import httpx
url, proc = http_server
if proc.poll() is not None:
pytest.skip("Server failed to start")
try:
response = httpx.get(f"{url}/health", timeout=5)
assert response.status_code == 200
data = response.json()
assert "status" in data
except httpx.ConnectError:
pytest.skip("Could not connect to server")
def test_airports_endpoint(self, http_server):
"""Test that airports endpoint responds."""
import httpx
url, proc = http_server
if proc.poll() is not None:
pytest.skip("Server failed to start")
try:
response = httpx.get(f"{url}/airports/by_city?city=Tokyo", timeout=5)
assert response.status_code == 200
except httpx.ConnectError:
pytest.skip("Could not connect to server")
class TestServerLifecycle:
"""Test server lifecycle and graceful shutdown."""
def test_server_handles_sigterm(self):
"""Test that server handles SIGTERM gracefully."""
import signal
proc = subprocess.Popen(
[sys.executable, "-m", "aerospace_mcp.fastmcp_server"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
time.sleep(0.3)
assert proc.poll() is None, "Server should be running"
# Send SIGTERM
proc.send_signal(signal.SIGTERM)
try:
proc.wait(timeout=5)
except subprocess.TimeoutExpired:
proc.kill()
pytest.fail("Server did not shut down gracefully")
def test_server_handles_stdin_close(self):
"""Test that server handles stdin close (client disconnect)."""
proc = subprocess.Popen(
[sys.executable, "-m", "aerospace_mcp.fastmcp_server"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
time.sleep(0.3)
assert proc.poll() is None, "Server should be running"
# Close stdin (simulate client disconnect)
proc.stdin.close()
# Server should exit or continue running (both acceptable)
try:
proc.wait(timeout=3)
except subprocess.TimeoutExpired:
# Server is still running, that's okay
proc.terminate()
proc.wait(timeout=5)
