mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-27 14:17:10 -04:00
87b54fc159
- Fix proxy crash when torch not installed: make kompress_compressor.py imports lazy so `is_kompress_available()` works without [ml] extra - Rewrite MCP server from 1 tool (retrieve-only) to 3 tools: headroom_compress (on-demand compression, no proxy needed), headroom_retrieve (local store first, proxy fallback), headroom_stats (session stats + sub-agent aggregation + proxy cache) - Add shared stats file (~/.headroom/session_stats.jsonl) so sub-agent compression stats are visible from the main session - Add mcp to [proxy] extras so proxy users get MCP tools automatically - Remove dead TextCompressor from exports and pipeline (was never called) - Update mcp install messaging to clarify proxy vs MCP roles - Fix fcntl Windows compat, asyncio deprecation, httpx timeout race Bump version to 0.4.6.
553 lines
21 KiB
Python
553 lines
21 KiB
Python
"""Integration tests for MCP CLI commands.
|
|
|
|
These are real tests that:
|
|
- Actually write/read config files
|
|
- Test actual CLI behavior
|
|
- Test MCP server initialization (when MCP SDK is available)
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from click.testing import CliRunner
|
|
|
|
from headroom.cli.main import main
|
|
from headroom.cli.mcp import (
|
|
get_headroom_command,
|
|
load_mcp_config,
|
|
save_mcp_config,
|
|
)
|
|
|
|
# Check if MCP SDK is available
|
|
try:
|
|
import mcp # noqa: F401
|
|
|
|
MCP_AVAILABLE = True
|
|
except ImportError:
|
|
MCP_AVAILABLE = False
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_claude_dir(tmp_path):
|
|
"""Create a temporary .claude directory for testing."""
|
|
claude_dir = tmp_path / ".claude"
|
|
claude_dir.mkdir()
|
|
return claude_dir
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_claude_config_path(temp_claude_dir):
|
|
"""Patch the MCP config path to use temp directory.
|
|
|
|
Also mocks the claude CLI as absent so tests exercise the mcp.json
|
|
fallback path rather than the `claude mcp add` path.
|
|
"""
|
|
import shutil as _shutil
|
|
|
|
config_path = temp_claude_dir / "mcp.json"
|
|
# Capture the original function reference before patching so we don't recurse.
|
|
_real_which = _shutil.which
|
|
|
|
def which_no_claude(cmd):
|
|
if cmd == "claude":
|
|
return None
|
|
return _real_which(cmd)
|
|
|
|
with patch("headroom.cli.mcp.MCP_CONFIG_PATH", config_path):
|
|
with patch("headroom.cli.mcp.CLAUDE_CONFIG_DIR", temp_claude_dir):
|
|
with patch("headroom.cli.mcp.shutil.which", side_effect=which_no_claude):
|
|
yield config_path
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_mcp_available():
|
|
"""Mock MCP SDK as available for testing install/uninstall commands."""
|
|
mock_mcp = MagicMock()
|
|
with patch.dict(sys.modules, {"mcp": mock_mcp}):
|
|
yield mock_mcp
|
|
|
|
|
|
class TestMCPConfigFunctions:
|
|
"""Test config file handling functions."""
|
|
|
|
def test_get_headroom_command_returns_list(self):
|
|
"""Command should be a list suitable for subprocess."""
|
|
cmd = get_headroom_command()
|
|
assert isinstance(cmd, list)
|
|
assert len(cmd) >= 1
|
|
# Should end with mcp serve args
|
|
assert "mcp" in cmd or "-m" in cmd
|
|
|
|
def test_load_mcp_config_empty_when_no_file(self, mock_claude_config_path):
|
|
"""Loading non-existent config returns empty structure."""
|
|
config = load_mcp_config()
|
|
assert config == {"mcpServers": {}}
|
|
|
|
def test_save_and_load_config(self, mock_claude_config_path):
|
|
"""Config can be saved and loaded back."""
|
|
test_config = {
|
|
"mcpServers": {
|
|
"headroom": {
|
|
"command": "headroom",
|
|
"args": ["mcp", "serve"],
|
|
}
|
|
}
|
|
}
|
|
save_mcp_config(test_config)
|
|
|
|
# File should exist
|
|
assert mock_claude_config_path.exists()
|
|
|
|
# Load it back
|
|
loaded = load_mcp_config()
|
|
assert loaded == test_config
|
|
|
|
def test_save_config_creates_directory(self, tmp_path):
|
|
"""save_mcp_config creates parent directory if needed."""
|
|
claude_dir = tmp_path / "new_dir" / ".claude"
|
|
config_path = claude_dir / "mcp.json"
|
|
|
|
with patch("headroom.cli.mcp.MCP_CONFIG_PATH", config_path):
|
|
with patch("headroom.cli.mcp.CLAUDE_CONFIG_DIR", claude_dir):
|
|
save_mcp_config({"mcpServers": {}})
|
|
|
|
assert config_path.exists()
|
|
|
|
def test_load_config_preserves_other_servers(self, mock_claude_config_path):
|
|
"""Loading preserves other MCP servers in config."""
|
|
# Write config with another server
|
|
existing_config = {
|
|
"mcpServers": {
|
|
"other-server": {"command": "other", "args": []},
|
|
}
|
|
}
|
|
mock_claude_config_path.write_text(json.dumps(existing_config))
|
|
|
|
loaded = load_mcp_config()
|
|
assert "other-server" in loaded["mcpServers"]
|
|
|
|
|
|
class TestMCPInstallCommand:
|
|
"""Test 'headroom mcp install' command."""
|
|
|
|
def test_install_creates_config(self, mock_claude_config_path, mock_mcp_available):
|
|
"""Install creates MCP config file."""
|
|
runner = CliRunner()
|
|
result = runner.invoke(main, ["mcp", "install"])
|
|
|
|
assert result.exit_code == 0, f"Failed with output: {result.output}"
|
|
assert "installed" in result.output.lower()
|
|
assert mock_claude_config_path.exists()
|
|
|
|
# Verify config content
|
|
config = json.loads(mock_claude_config_path.read_text())
|
|
assert "headroom" in config["mcpServers"]
|
|
assert config["mcpServers"]["headroom"]["command"] == "headroom"
|
|
assert "mcp" in config["mcpServers"]["headroom"]["args"]
|
|
assert "serve" in config["mcpServers"]["headroom"]["args"]
|
|
|
|
def test_install_preserves_other_servers(self, mock_claude_config_path, mock_mcp_available):
|
|
"""Install preserves existing MCP servers."""
|
|
# Create config with another server
|
|
existing_config = {
|
|
"mcpServers": {
|
|
"github": {"command": "github-mcp", "args": []},
|
|
}
|
|
}
|
|
mock_claude_config_path.write_text(json.dumps(existing_config))
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(main, ["mcp", "install"])
|
|
|
|
assert result.exit_code == 0, f"Failed with output: {result.output}"
|
|
|
|
# Both servers should exist
|
|
config = json.loads(mock_claude_config_path.read_text())
|
|
assert "github" in config["mcpServers"]
|
|
assert "headroom" in config["mcpServers"]
|
|
|
|
def test_install_with_custom_proxy_url(self, mock_claude_config_path, mock_mcp_available):
|
|
"""Install with custom proxy URL sets env var."""
|
|
runner = CliRunner()
|
|
result = runner.invoke(main, ["mcp", "install", "--proxy-url", "http://localhost:9000"])
|
|
|
|
assert result.exit_code == 0, f"Failed with output: {result.output}"
|
|
|
|
config = json.loads(mock_claude_config_path.read_text())
|
|
assert (
|
|
config["mcpServers"]["headroom"]["env"]["HEADROOM_PROXY_URL"] == "http://localhost:9000"
|
|
)
|
|
|
|
def test_install_default_proxy_url_no_env(self, mock_claude_config_path, mock_mcp_available):
|
|
"""Install with default proxy URL doesn't set env var."""
|
|
runner = CliRunner()
|
|
result = runner.invoke(main, ["mcp", "install"])
|
|
|
|
assert result.exit_code == 0, f"Failed with output: {result.output}"
|
|
|
|
config = json.loads(mock_claude_config_path.read_text())
|
|
# No env section for default URL
|
|
assert "env" not in config["mcpServers"]["headroom"]
|
|
|
|
def test_install_already_configured_no_force(self, mock_claude_config_path, mock_mcp_available):
|
|
"""Install without --force when already configured exits cleanly."""
|
|
# First install
|
|
runner = CliRunner()
|
|
runner.invoke(main, ["mcp", "install"])
|
|
|
|
# Second install without force
|
|
result = runner.invoke(main, ["mcp", "install"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "already configured" in result.output.lower()
|
|
|
|
def test_install_force_overwrites(self, mock_claude_config_path, mock_mcp_available):
|
|
"""Install with --force overwrites existing config."""
|
|
runner = CliRunner()
|
|
runner.invoke(main, ["mcp", "install", "--proxy-url", "http://old:8787"])
|
|
|
|
# Force install with new URL
|
|
result = runner.invoke(
|
|
main, ["mcp", "install", "--force", "--proxy-url", "http://new:9000"]
|
|
)
|
|
|
|
assert result.exit_code == 0, f"Failed with output: {result.output}"
|
|
assert "installed" in result.output.lower()
|
|
|
|
config = json.loads(mock_claude_config_path.read_text())
|
|
assert config["mcpServers"]["headroom"]["env"]["HEADROOM_PROXY_URL"] == "http://new:9000"
|
|
|
|
@pytest.mark.skipif(MCP_AVAILABLE, reason="Test only runs when MCP SDK is NOT installed")
|
|
def test_install_without_mcp_sdk_fails(self, mock_claude_config_path):
|
|
"""Install fails gracefully when MCP SDK is not installed."""
|
|
runner = CliRunner()
|
|
result = runner.invoke(main, ["mcp", "install"])
|
|
|
|
# Should fail with helpful message
|
|
assert result.exit_code == 1
|
|
assert "mcp" in result.output.lower() or "not installed" in result.output.lower()
|
|
|
|
|
|
class TestMCPUninstallCommand:
|
|
"""Test 'headroom mcp uninstall' command."""
|
|
|
|
def test_uninstall_removes_headroom(self, mock_claude_config_path, mock_mcp_available):
|
|
"""Uninstall removes headroom from config."""
|
|
# First install
|
|
runner = CliRunner()
|
|
runner.invoke(main, ["mcp", "install"])
|
|
|
|
# Then uninstall
|
|
result = runner.invoke(main, ["mcp", "uninstall"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "removed" in result.output.lower()
|
|
|
|
config = json.loads(mock_claude_config_path.read_text())
|
|
assert "headroom" not in config["mcpServers"]
|
|
|
|
def test_uninstall_preserves_other_servers(self, mock_claude_config_path):
|
|
"""Uninstall preserves other MCP servers."""
|
|
# Create config with headroom and another server
|
|
config = {
|
|
"mcpServers": {
|
|
"headroom": {"command": "headroom", "args": ["mcp", "serve"]},
|
|
"github": {"command": "github-mcp", "args": []},
|
|
}
|
|
}
|
|
mock_claude_config_path.write_text(json.dumps(config))
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(main, ["mcp", "uninstall"])
|
|
|
|
assert result.exit_code == 0
|
|
|
|
config = json.loads(mock_claude_config_path.read_text())
|
|
assert "headroom" not in config["mcpServers"]
|
|
assert "github" in config["mcpServers"]
|
|
|
|
def test_uninstall_no_config_file(self, mock_claude_config_path):
|
|
"""Uninstall with no config file exits cleanly."""
|
|
runner = CliRunner()
|
|
result = runner.invoke(main, ["mcp", "uninstall"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "nothing to uninstall" in result.output.lower()
|
|
|
|
def test_uninstall_not_configured(self, mock_claude_config_path):
|
|
"""Uninstall when headroom not in config exits cleanly."""
|
|
# Create config without headroom
|
|
config = {"mcpServers": {"other": {"command": "other"}}}
|
|
mock_claude_config_path.write_text(json.dumps(config))
|
|
|
|
runner = CliRunner()
|
|
result = runner.invoke(main, ["mcp", "uninstall"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "not configured" in result.output.lower()
|
|
|
|
|
|
class TestMCPStatusCommand:
|
|
"""Test 'headroom mcp status' command."""
|
|
|
|
def test_status_not_configured(self, mock_claude_config_path):
|
|
"""Status shows not configured when no config."""
|
|
runner = CliRunner()
|
|
result = runner.invoke(main, ["mcp", "status"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "MCP SDK" in result.output
|
|
# Should show not configured
|
|
assert (
|
|
"✗" in result.output
|
|
or "Not configured" in result.output.lower()
|
|
or "No config" in result.output
|
|
)
|
|
|
|
def test_status_configured(self, mock_claude_config_path, mock_mcp_available):
|
|
"""Status shows configured when installed."""
|
|
runner = CliRunner()
|
|
runner.invoke(main, ["mcp", "install"])
|
|
|
|
result = runner.invoke(main, ["mcp", "status"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "✓ Configured" in result.output
|
|
|
|
|
|
class TestMCPServeCommand:
|
|
"""Test 'headroom mcp serve' command."""
|
|
|
|
def test_serve_help(self):
|
|
"""Serve command shows help."""
|
|
runner = CliRunner()
|
|
result = runner.invoke(main, ["mcp", "serve", "--help"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "proxy-url" in result.output
|
|
assert "debug" in result.output
|
|
|
|
|
|
@pytest.mark.skipif(not MCP_AVAILABLE, reason="MCP SDK not installed")
|
|
class TestMCPServerInitialization:
|
|
"""Test actual MCP server creation.
|
|
|
|
These tests require the MCP SDK to be installed.
|
|
"""
|
|
|
|
def test_mcp_server_can_be_created(self):
|
|
"""MCP server can be instantiated."""
|
|
from headroom.ccr.mcp_server import create_ccr_mcp_server
|
|
|
|
server = create_ccr_mcp_server()
|
|
assert server is not None
|
|
assert server.proxy_url == "http://127.0.0.1:8787"
|
|
|
|
def test_mcp_server_with_custom_url(self):
|
|
"""MCP server accepts custom proxy URL."""
|
|
from headroom.ccr.mcp_server import create_ccr_mcp_server
|
|
|
|
server = create_ccr_mcp_server(proxy_url="http://custom:9000")
|
|
assert server.proxy_url == "http://custom:9000"
|
|
|
|
def test_mcp_server_has_correct_tool_name(self):
|
|
"""MCP server is configured for headroom_retrieve tool."""
|
|
from headroom.ccr.mcp_server import create_ccr_mcp_server
|
|
from headroom.ccr.tool_injection import CCR_TOOL_NAME
|
|
|
|
server = create_ccr_mcp_server()
|
|
|
|
# Verify the server was created with correct configuration
|
|
assert server.server is not None
|
|
assert server.server.name == "headroom"
|
|
# The tool name should be headroom_retrieve
|
|
assert CCR_TOOL_NAME == "headroom_retrieve"
|
|
|
|
|
|
class TestEndToEndFlow:
|
|
"""Test complete install -> status -> uninstall flow."""
|
|
|
|
def test_full_lifecycle(self, mock_claude_config_path, mock_mcp_available):
|
|
"""Test complete lifecycle of MCP configuration."""
|
|
runner = CliRunner()
|
|
|
|
# Initially not configured
|
|
result = runner.invoke(main, ["mcp", "status"])
|
|
assert "No config" in result.output or "Not configured" in result.output.lower()
|
|
|
|
# Install
|
|
result = runner.invoke(main, ["mcp", "install"])
|
|
assert result.exit_code == 0, f"Install failed: {result.output}"
|
|
assert "installed" in result.output.lower()
|
|
|
|
# Status shows configured
|
|
result = runner.invoke(main, ["mcp", "status"])
|
|
assert "✓ Configured" in result.output
|
|
|
|
# Config file has correct content
|
|
config = json.loads(mock_claude_config_path.read_text())
|
|
assert config["mcpServers"]["headroom"]["command"] == "headroom"
|
|
|
|
# Uninstall
|
|
result = runner.invoke(main, ["mcp", "uninstall"])
|
|
assert result.exit_code == 0
|
|
assert "removed" in result.output.lower()
|
|
|
|
# Status shows not configured
|
|
result = runner.invoke(main, ["mcp", "status"])
|
|
assert "headroom" not in result.output.lower() or "not configured" in result.output.lower()
|
|
|
|
|
|
class TestMCPInstallWithClaudeCLI:
|
|
"""Test mcp_install when the claude CLI is available."""
|
|
|
|
def _make_run(self, get_rc=1, add_rc=0, remove_rc=0):
|
|
"""Return a subprocess.run mock with configurable return codes."""
|
|
|
|
def run(cmd, **kwargs):
|
|
if "get" in cmd:
|
|
return MagicMock(returncode=get_rc, stderr="")
|
|
if "remove" in cmd:
|
|
return MagicMock(returncode=remove_rc, stderr="")
|
|
if "add" in cmd:
|
|
return MagicMock(returncode=add_rc, stderr="")
|
|
return MagicMock(returncode=0, stderr="")
|
|
|
|
return run
|
|
|
|
def test_install_uses_claude_mcp_add(self, mock_mcp_available):
|
|
"""When claude CLI is available, install calls claude mcp add."""
|
|
runner = CliRunner()
|
|
with patch("headroom.cli.mcp.shutil.which", return_value="/usr/bin/claude"):
|
|
with patch("headroom.cli.mcp.subprocess.run", side_effect=self._make_run()) as mock_run:
|
|
result = runner.invoke(main, ["mcp", "install"])
|
|
|
|
assert result.exit_code == 0, f"Failed: {result.output}"
|
|
assert "installed" in result.output.lower()
|
|
assert "claude mcp add" in result.output
|
|
|
|
# Verify claude mcp add was called
|
|
add_calls = [c for c in mock_run.call_args_list if "add" in c.args[0]]
|
|
assert len(add_calls) == 1
|
|
add_cmd = add_calls[0].args[0]
|
|
assert "headroom" in add_cmd
|
|
assert "-s" in add_cmd
|
|
assert "user" in add_cmd
|
|
|
|
def test_install_already_registered_no_force(self, mock_mcp_available):
|
|
"""Install without --force exits cleanly when already registered via claude CLI."""
|
|
runner = CliRunner()
|
|
with patch("headroom.cli.mcp.shutil.which", return_value="/usr/bin/claude"):
|
|
# get returns 0 → already registered
|
|
with patch("headroom.cli.mcp.subprocess.run", side_effect=self._make_run(get_rc=0)):
|
|
result = runner.invoke(main, ["mcp", "install"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "already configured" in result.output.lower()
|
|
|
|
def test_install_force_calls_remove_then_add(self, mock_mcp_available):
|
|
"""--force calls claude mcp remove before claude mcp add."""
|
|
runner = CliRunner()
|
|
calls = []
|
|
|
|
def capturing_run(cmd, **kwargs):
|
|
calls.append(cmd)
|
|
return MagicMock(returncode=0, stderr="")
|
|
|
|
with patch("headroom.cli.mcp.shutil.which", return_value="/usr/bin/claude"):
|
|
with patch("headroom.cli.mcp.subprocess.run", side_effect=capturing_run):
|
|
result = runner.invoke(main, ["mcp", "install", "--force"])
|
|
|
|
assert result.exit_code == 0, f"Failed: {result.output}"
|
|
subcommands = [c[2] for c in calls] # third element is the subcommand
|
|
assert "remove" in subcommands
|
|
assert "add" in subcommands
|
|
assert subcommands.index("remove") < subcommands.index("add")
|
|
|
|
def test_install_fallback_on_claude_mcp_add_failure(self, temp_claude_dir, mock_mcp_available):
|
|
"""If claude mcp add fails, falls back to writing mcp.json."""
|
|
config_path = temp_claude_dir / "mcp.json"
|
|
with patch("headroom.cli.mcp.MCP_CONFIG_PATH", config_path):
|
|
with patch("headroom.cli.mcp.CLAUDE_CONFIG_DIR", temp_claude_dir):
|
|
with patch("headroom.cli.mcp.shutil.which", return_value="/usr/bin/claude"):
|
|
with patch(
|
|
"headroom.cli.mcp.subprocess.run",
|
|
side_effect=self._make_run(add_rc=1),
|
|
):
|
|
runner = CliRunner()
|
|
result = runner.invoke(main, ["mcp", "install"])
|
|
|
|
assert result.exit_code == 0, f"Failed: {result.output}"
|
|
assert config_path.exists()
|
|
config = json.loads(config_path.read_text())
|
|
assert "headroom" in config["mcpServers"]
|
|
|
|
def test_install_with_custom_proxy_url_passes_e_flag(self, mock_mcp_available):
|
|
"""Custom proxy URL is passed as -e KEY=VALUE to claude mcp add."""
|
|
calls = []
|
|
|
|
def capturing_run(cmd, **kwargs):
|
|
calls.append(list(cmd))
|
|
# Return rc=1 for "get" so headroom is treated as not yet registered
|
|
if "get" in cmd:
|
|
return MagicMock(returncode=1, stderr="")
|
|
return MagicMock(returncode=0, stderr="")
|
|
|
|
runner = CliRunner()
|
|
with patch("headroom.cli.mcp.shutil.which", return_value="/usr/bin/claude"):
|
|
with patch("headroom.cli.mcp.subprocess.run", side_effect=capturing_run):
|
|
result = runner.invoke(
|
|
main, ["mcp", "install", "--proxy-url", "http://custom:9000"]
|
|
)
|
|
|
|
assert result.exit_code == 0, f"Failed: {result.output}"
|
|
add_calls = [c for c in calls if "add" in c]
|
|
assert len(add_calls) == 1
|
|
add_cmd = add_calls[0]
|
|
assert "-e" in add_cmd
|
|
env_idx = add_cmd.index("-e")
|
|
assert add_cmd[env_idx + 1] == "HEADROOM_PROXY_URL=http://custom:9000"
|
|
|
|
|
|
class TestMCPUninstallWithClaudeCLI:
|
|
"""Test mcp_uninstall when the claude CLI is available."""
|
|
|
|
def test_uninstall_calls_claude_mcp_remove(self):
|
|
"""Uninstall calls claude mcp remove when headroom is registered."""
|
|
calls = []
|
|
|
|
def capturing_run(cmd, **kwargs):
|
|
calls.append(list(cmd))
|
|
return MagicMock(returncode=0, stderr="")
|
|
|
|
runner = CliRunner()
|
|
with patch("headroom.cli.mcp.shutil.which", return_value="/usr/bin/claude"):
|
|
with patch("headroom.cli.mcp.subprocess.run", side_effect=capturing_run):
|
|
result = runner.invoke(main, ["mcp", "uninstall"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "removed" in result.output.lower()
|
|
subcommands = [c[2] for c in calls]
|
|
assert "remove" in subcommands
|
|
|
|
def test_uninstall_skips_remove_when_not_registered(self):
|
|
"""Uninstall does not call remove when headroom is not registered via claude CLI."""
|
|
calls = []
|
|
|
|
def capturing_run(cmd, **kwargs):
|
|
calls.append(list(cmd))
|
|
# mcp get returns non-zero → not registered
|
|
if "get" in cmd:
|
|
return MagicMock(returncode=1, stderr="")
|
|
return MagicMock(returncode=0, stderr="")
|
|
|
|
runner = CliRunner()
|
|
with patch("headroom.cli.mcp.shutil.which", return_value="/usr/bin/claude"):
|
|
with patch("headroom.cli.mcp.subprocess.run", side_effect=capturing_run):
|
|
result = runner.invoke(main, ["mcp", "uninstall"])
|
|
|
|
assert result.exit_code == 0
|
|
subcommands = [c[2] for c in calls]
|
|
assert "remove" not in subcommands
|