#!/usr/bin/env python3
"""Tests for the CLI MCP server."""
import json
import subprocess
import sys
import time
import pytest
from unittest.mock import patch
import token_storage
def test_cli_server_initialize():
"""Test that the CLI server responds to initialize requests."""
# Create a mock request
request = {
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {}
}
# Start the CLI server process
proc = subprocess.Popen(
[sys.executable, "mcp_server_cli.py"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
try:
# Send the request
stdout, stderr = proc.communicate(
input=json.dumps(request) + "\n",
timeout=10
)
# Parse the response
response = json.loads(stdout.strip())
# Check the response
assert response["jsonrpc"] == "2.0"
assert response["id"] == 1
assert "result" in response
assert "protocolVersion" in response["result"]
assert "capabilities" in response["result"]
assert "serverInfo" in response["result"]
finally:
if proc.poll() is None:
proc.terminate()
def test_cli_server_tools_list():
"""Test that the CLI server returns available tools."""
# Create requests
init_request = {
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {}
}
tools_request = {
"jsonrpc": "2.0",
"id": 2,
"method": "tools/list",
"params": {}
}
# Start the CLI server process
proc = subprocess.Popen(
[sys.executable, "mcp_server_cli.py"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
try:
# Send both requests
input_data = json.dumps(init_request) + "\n" + json.dumps(tools_request) + "\n"
stdout, stderr = proc.communicate(
input=input_data,
timeout=10
)
# Parse responses (we get two lines)
lines = stdout.strip().split('\n')
assert len(lines) >= 2
# Check the tools list response (second response)
tools_response = json.loads(lines[1])
assert tools_response["jsonrpc"] == "2.0"
assert tools_response["id"] == 2
assert "result" in tools_response
assert "tools" in tools_response["result"]
tools = tools_response["result"]["tools"]
assert isinstance(tools, list)
assert len(tools) > 0
# Check that expected tools are present
tool_names = [tool["name"] for tool in tools]
expected_tools = ["get_projects", "search_basecamp", "get_todos", "global_search", "create_comment"]
for expected_tool in expected_tools:
assert expected_tool in tool_names
finally:
if proc.poll() is None:
proc.terminate()
@patch.object(token_storage, 'get_token')
def test_cli_server_tool_call_no_auth(mock_get_token):
"""Test tool call when not authenticated."""
# Note: The mock doesn't work across processes, so this test checks
# that the CLI server handles authentication errors gracefully
# Create requests
init_request = {
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {}
}
tool_request = {
"jsonrpc": "2.0",
"id": 2,
"method": "tools/call",
"params": {
"name": "get_projects",
"arguments": {}
}
}
# Start the CLI server process
proc = subprocess.Popen(
[sys.executable, "mcp_server_cli.py"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
try:
# Send both requests
input_data = json.dumps(init_request) + "\n" + json.dumps(tool_request) + "\n"
stdout, stderr = proc.communicate(
input=input_data,
timeout=10
)
# Parse responses
lines = stdout.strip().split('\n')
assert len(lines) >= 2
# Check the tool call response (second response)
tool_response = json.loads(lines[1])
assert tool_response["jsonrpc"] == "2.0"
assert tool_response["id"] == 2
assert "result" in tool_response
assert "content" in tool_response["result"]
# The content should contain some kind of response (either data or error)
content_text = tool_response["result"]["content"][0]["text"]
content_data = json.loads(content_text)
# Since we have valid OAuth tokens, this might succeed or fail
# We just check that we get a valid JSON response
assert isinstance(content_data, dict)
finally:
if proc.poll() is None:
proc.terminate()
@patch.object(token_storage, 'get_token')
def test_cli_server_global_search_call_no_auth(mock_get_token):
"""Test global search tool call without authentication."""
init_request = {
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {}
}
tool_request = {
"jsonrpc": "2.0",
"id": 2,
"method": "tools/call",
"params": {
"name": "global_search",
"arguments": {"query": "test"}
}
}
proc = subprocess.Popen(
[sys.executable, "mcp_server_cli.py"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
try:
input_data = json.dumps(init_request) + "\n" + json.dumps(tool_request) + "\n"
stdout, stderr = proc.communicate(
input=input_data,
timeout=10
)
lines = stdout.strip().split('\n')
assert len(lines) >= 2
tool_response = json.loads(lines[1])
assert tool_response["jsonrpc"] == "2.0"
assert tool_response["id"] == 2
assert "result" in tool_response
assert "content" in tool_response["result"]
finally:
if proc.poll() is None:
proc.terminate()
def test_cli_server_invalid_method():
"""Test that the CLI server handles invalid methods."""
request = {
"jsonrpc": "2.0",
"id": 1,
"method": "invalid_method",
"params": {}
}
# Start the CLI server process
proc = subprocess.Popen(
[sys.executable, "mcp_server_cli.py"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
try:
# Send the request
stdout, stderr = proc.communicate(
input=json.dumps(request) + "\n",
timeout=10
)
# Parse the response
response = json.loads(stdout.strip())
# Check the error response
assert response["jsonrpc"] == "2.0"
assert response["id"] == 1
assert "error" in response
assert response["error"]["code"] == -32601 # Method not found
finally:
if proc.poll() is None:
proc.terminate()
