#!/usr/bin/env python3
"""
Integration tests for API Manager MCP Server
"""
import os
import tempfile
import pytest
import asyncio
import json
from pathlib import Path
from unittest.mock import patch, MagicMock, AsyncMock
# Import the main module
import sys
sys.path.insert(0, str(Path(__file__).parent.parent))
from api_manager import APIKeyManager, call_tool
class TestAPIKeyManagerIntegration:
"""Integration tests for the complete API Manager workflow"""
def setup_method(self):
"""Setup test environment"""
self.temp_dir = tempfile.mkdtemp()
self.test_env_file = Path(self.temp_dir) / ".env"
# Patch the global manager to use our test file
from api_manager import manager
manager.env_file = self.test_env_file
manager._add_initial_header()
def teardown_method(self):
"""Cleanup"""
import shutil
shutil.rmtree(self.temp_dir, ignore_errors=True)
@pytest.mark.asyncio
async def test_complete_workflow(self):
"""Test complete workflow: add, list, search, delete"""
# 1. Add API keys
add_result = await call_tool("add_api_key", {
"key_name": "OPENAI_API_KEY",
"value": "sk-1234567890abcdef1234567890abcdef"
})
assert "Successfully added" in add_result[0].text
add_result2 = await call_tool("add_api_key", {
"key_name": "ANTHROPIC_API_KEY",
"value": "sk-ant-1234567890abcdef1234567890abcdef"
})
assert "Successfully added" in add_result2[0].text
# 2. List all keys
list_result = await call_tool("list_api_keys", {"show_values": False})
assert "OpenAI" in list_result[0].text
assert "Anthropic" in list_result[0].text
assert "Total API Keys: 2" in list_result[0].text
# 3. Search for keys
search_result = await call_tool("search_api_keys", {
"query": "openai",
"show_values": False
})
assert "OPENAI_API_KEY" in search_result[0].text
assert "Found 1 matching key(s)" in search_result[0].text
# 4. Get specific key
get_result = await call_tool("get_api_key", {
"key_name": "OPENAI_API_KEY"
})
assert "sk-1234567890abcdef1234567890abcdef" in get_result[0].text
# 5. Get statistics
stats_result = await call_tool("get_stats", {})
assert "Total API Keys: 2" in stats_result[0].text
assert "OpenAI" in stats_result[0].text
# 6. Create backup
backup_result = await call_tool("backup_env_file", {})
assert "Backup created successfully" in backup_result[0].text
# 7. Export keys
export_result = await call_tool("export_keys", {
"format": "json",
"include_values": False
})
assert "OPENAI_API_KEY" in export_result[0].text
# 8. Delete key
delete_result = await call_tool("delete_api_key", {
"key_name": "OPENAI_API_KEY"
})
assert "Successfully deleted" in delete_result[0].text
# 9. Verify deletion
final_list = await call_tool("list_api_keys", {})
assert "Total API Keys: 1" in final_list[0].text
@pytest.mark.asyncio
async def test_error_handling(self):
"""Test error handling in various scenarios"""
# Test invalid key name
invalid_name_result = await call_tool("add_api_key", {
"key_name": "invalid/key*name",
"value": "valid-key-value-123"
})
assert "Invalid key name" in invalid_name_result[0].text
# Test invalid key value
invalid_value_result = await call_tool("add_api_key", {
"key_name": "VALID_KEY_NAME",
"value": "test"
})
assert "Invalid key value" in invalid_value_result[0].text
# Test getting non-existent key
missing_key_result = await call_tool("get_api_key", {
"key_name": "NONEXISTENT_KEY"
})
assert "not found" in missing_key_result[0].text
# Test deleting non-existent key
delete_missing_result = await call_tool("delete_api_key", {
"key_name": "NONEXISTENT_KEY"
})
assert "not found" in delete_missing_result[0].text
# Test invalid export format
invalid_export_result = await call_tool("export_keys", {
"format": "invalid_format"
})
assert "Export Error" in invalid_export_result[0].text
@pytest.mark.asyncio
async def test_category_filtering(self):
"""Test category-based filtering"""
# Add keys from different categories
keys_to_add = [
("OPENAI_API_KEY", "sk-openai123", "OpenAI"),
("ANTHROPIC_API_KEY", "sk-ant-anthropic123", "Anthropic (Claude)"),
("DISCORD_BOT_TOKEN", "discord-token-123", "Discord"),
("GITHUB_TOKEN", "github-token-123", "GitHub")
]
for key_name, value, expected_category in keys_to_add:
await call_tool("add_api_key", {"key_name": key_name, "value": value})
# Test listing by category
ai_keys_result = await call_tool("list_api_keys", {"category": "OpenAI"})
assert "OPENAI_API_KEY" in ai_keys_result[0].text
assert "DISCORD_BOT_TOKEN" not in ai_keys_result[0].text
# Test list categories
categories_result = await call_tool("list_categories", {})
assert "OpenAI" in categories_result[0].text
assert "Discord" in categories_result[0].text
@pytest.mark.asyncio
async def test_security_features(self):
"""Test security-related features"""
# Add a test key
await call_tool("add_api_key", {
"key_name": "SECRET_KEY",
"value": "very-secret-key-12345678901234567890"
})
# Test that values are masked by default
list_result = await call_tool("list_api_keys", {})
assert "very-secret-key-12345678901234567890" not in list_result[0].text
assert "very****" in list_result[0].text
# Test that search results are masked by default
search_result = await call_tool("search_api_keys", {"query": "SECRET"})
assert "very-secret-key-12345678901234567890" not in search_result[0].text
# Test showing values when explicitly requested
list_with_values = await call_tool("list_api_keys", {"show_values": True})
assert "very-secret-key-12345678901234567890" in list_with_values[0].text
@pytest.mark.asyncio
async def test_backup_and_restore_workflow(self):
"""Test backup and restore functionality"""
# Add initial keys
await call_tool("add_api_key", {
"key_name": "BACKUP_TEST_KEY",
"value": "backup-test-value-123"
})
# Create backup
backup_result = await call_tool("backup_env_file", {})
assert "Backup created successfully" in backup_result[0].text
# Delete the key (this should also create a backup)
delete_result = await call_tool("delete_api_key", {
"key_name": "BACKUP_TEST_KEY"
})
assert "Successfully deleted" in delete_result[0].text
assert "Backup created" in delete_result[0].text
# Verify key is gone
list_result = await call_tool("list_api_keys", {})
assert "BACKUP_TEST_KEY" not in list_result[0].text
@pytest.mark.asyncio
async def test_update_existing_key(self):
"""Test updating an existing key"""
# Add initial key
await call_tool("add_api_key", {
"key_name": "UPDATE_TEST_KEY",
"value": "original-value-123456"
})
# Update the same key
update_result = await call_tool("add_api_key", {
"key_name": "UPDATE_TEST_KEY",
"value": "updated-value-789012"
})
assert "Successfully updated" in update_result[0].text
# Verify the update
get_result = await call_tool("get_api_key", {
"key_name": "UPDATE_TEST_KEY"
})
assert "updated-value-789012" in get_result[0].text
@pytest.mark.asyncio
async def test_large_dataset_performance(self):
"""Test performance with larger datasets"""
# Add many keys
for i in range(50):
await call_tool("add_api_key", {
"key_name": f"TEST_KEY_{i:03d}",
"value": f"test-value-{i:03d}-{'x' * 20}"
})
# Test listing performance
list_result = await call_tool("list_api_keys", {})
assert "Total API Keys: 50" in list_result[0].text
# Test search performance
search_result = await call_tool("search_api_keys", {
"query": "TEST_KEY_025"
})
assert "Found 1 matching key(s)" in search_result[0].text
# Test statistics generation
stats_result = await call_tool("get_stats", {})
assert "Total API Keys: 50" in stats_result[0].text
@pytest.mark.asyncio
async def test_concurrent_operations(self):
"""Test concurrent operations safety"""
# Simulate concurrent adds
tasks = []
for i in range(10):
task = call_tool("add_api_key", {
"key_name": f"CONCURRENT_KEY_{i}",
"value": f"concurrent-value-{i}-{'x' * 20}"
})
tasks.append(task)
# Wait for all operations to complete
results = await asyncio.gather(*tasks)
# Verify all operations succeeded
for result in results:
assert "Successfully added" in result[0].text
# Verify all keys are present
list_result = await call_tool("list_api_keys", {})
assert "Total API Keys: 10" in list_result[0].text
@pytest.mark.asyncio
async def test_export_import_workflow(self):
"""Test export and import workflow"""
# Add test data
test_keys = {
"EXPORT_TEST_1": "export-value-1-abcdef",
"EXPORT_TEST_2": "export-value-2-123456"
}
for key_name, value in test_keys.items():
await call_tool("add_api_key", {
"key_name": key_name,
"value": value
})
# Test JSON export with values
json_result = await call_tool("export_keys", {
"format": "json",
"include_values": True
})
assert "EXPORT_TEST_1" in json_result[0].text
assert "export-value-1-abcdef" in json_result[0].text
# Test ENV export without values
env_result = await call_tool("export_keys", {
"format": "env",
"include_values": False
})
assert "EXPORT_TEST_1=" in env_result[0].text
assert "export-value-1-abcdef" not in env_result[0].text
@pytest.mark.asyncio
async def test_malformed_input_handling(self):
"""Test handling of malformed input"""
# Test with missing required parameters
try:
await call_tool("add_api_key", {"key_name": "TEST_KEY"})
except Exception:
pass # Expected to fail
try:
await call_tool("get_api_key", {})
except Exception:
pass # Expected to fail
# Test with unexpected parameter types
result = await call_tool("list_api_keys", {"show_values": "not_a_boolean"})
# Should handle gracefully
if __name__ == "__main__":
pytest.main([__file__])
