Initial commit: MCP Hub Community Edition v3.0.0

Community edition generated from private repo via sync pipeline.
Includes 9 plugins (WordPress, WooCommerce, WP Advanced, Gitea, n8n,
Supabase, OpenPanel, Appwrite, Directus) with ~587 tools.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
airano
2026-02-17 08:34:44 +03:30
commit cf62e65c55
237 changed files with 87596 additions and 0 deletions

115
tests/README.md Normal file
View File

@@ -0,0 +1,115 @@
# Testing Guide
## Integration Tests
### Quick Start
```bash
# In Docker container:
docker exec -it mcphub python tests/run_integration_tests.py
# Or local:
python tests/run_integration_tests.py
```
### What Gets Tested
1. **Module Imports** - All modules load correctly
2. **BasePlugin Architecture** - Plugin system compatibility
3. **WordPress Plugin** - Initialization and handlers
4. **Tool Specifications** - Format and completeness
5. **Handlers Structure** - All 14 handlers present
6. **Pydantic Schemas** - Validation working
7. **ToolGenerator Integration** - Dynamic tool generation ready
### Expected Output
```
Running Integration Tests for WordPress MCP Server
============================================================
Running: Imports... PASSED
Running: Base Plugin... PASSED
Running: WordPress Plugin Init... PASSED
Running: Tool Specifications... PASSED
Running: Handlers Structure... PASSED
Running: Pydantic Schemas... PASSED
Running: ToolGenerator Integration... PASSED
============================================================
Test Summary:
Total: 7
Passed: 7
Failed: 0
Skipped: 0
```
### Exit Codes
- `0` - All tests passed
- `1` - Some tests failed
### JSON Output
The script also outputs results as JSON for programmatic parsing:
```json
{
"timestamp": "2025-11-16T10:30:00",
"tests": [
{
"name": "imports",
"status": "passed",
"message": "All modules imported successfully",
"details": {}
}
],
"summary": {
"total": 7,
"passed": 7,
"failed": 0,
"skipped": 0
}
}
```
## Manual Testing with Real Sites
To test with real sites, use MCP Inspector or Claude Desktop:
```python
# Test health
wordpress_get_site_health(site="mysite")
# Test posts
wordpress_list_posts(site="mysite", per_page=5)
# Test products
wordpress_list_products(site="mysite", per_page=5)
# Test WP-CLI (requires container access)
wordpress_wp_cache_type(site="mysite")
```
## Troubleshooting
### Import Errors
```
Error: ModuleNotFoundError: No module named 'plugins'
```
**Fix**: Run from project root:
```bash
cd /app # In Docker
python tests/run_integration_tests.py
```
### Handler Errors
```
Error: Can't instantiate abstract class WordPressPlugin
```
**Fix**: This means BasePlugin still has abstract get_tools().
This should be fixed in the latest version.

View File

@@ -0,0 +1,143 @@
#!/usr/bin/env python3
"""
Test API Key Access Control Fix
Validates that:
1. Site-level keys get proper error message for system tools (not validation failure)
2. Global keys work for all tools
3. Per-project keys work for their own unified tools
"""
def test_system_tool_detection():
"""Test that system tools are detected correctly."""
SYSTEM_TOOLS = [
"list_projects",
"get_project_info",
"check_all_projects_health",
"get_project_health",
"get_system_metrics",
"get_system_uptime",
"get_rate_limit_stats",
"export_health_metrics",
"manage_api_keys_list",
"manage_api_keys_get_info",
]
# Test with and without MCP prefix
test_cases = [
("list_projects", True),
("mcp__coolify-projects__list_projects", True),
("get_project_info", True),
("wordpress_list_posts", False),
("woocommerce_list_orders", False),
("unknown", False),
]
for tool_name, expected in test_cases:
is_system_tool = any(tool_name.endswith(st) for st in SYSTEM_TOOLS)
assert (
is_system_tool == expected
), f"Failed for {tool_name}: expected {expected}, got {is_system_tool}"
print(f"{tool_name}: is_system_tool={is_system_tool}")
def test_unified_tool_detection():
"""Test that unified tools are detected correctly."""
test_cases = [
("wordpress_list_posts", True),
("woocommerce_list_orders", True),
("mcp__coolify-projects__wordpress_list_posts", True),
("mcp__coolify-projects__woocommerce_list_orders", True),
("list_projects", False),
("get_project_info", False),
("unknown", True), # Falls to fallback
]
for tool_name, expected in test_cases:
if tool_name == "unknown":
is_unified_tool = True # Fallback
else:
is_unified_tool = (
tool_name.startswith("wordpress_")
or tool_name.startswith("woocommerce_")
or tool_name.startswith("mcp__coolify-projects__wordpress_")
or tool_name.startswith("mcp__coolify-projects__woocommerce_")
)
assert (
is_unified_tool == expected
), f"Failed for {tool_name}: expected {expected}, got {is_unified_tool}"
print(f"{tool_name}: is_unified_tool={is_unified_tool}")
def test_skip_project_check_logic():
"""Test that skip_project_check is set correctly."""
SYSTEM_TOOLS = [
"list_projects",
"get_project_info",
"check_all_projects_health",
"get_project_health",
"get_system_metrics",
"get_system_uptime",
"get_rate_limit_stats",
"export_health_metrics",
"manage_api_keys_list",
"manage_api_keys_get_info",
]
test_cases = [
# (tool_name, expected_skip)
("list_projects", True), # System tool
("wordpress_list_posts", True), # Unified tool
("woocommerce_list_orders", True), # Unified tool
("mcp__coolify-projects__list_projects", True), # System tool with prefix
("mcp__coolify-projects__wordpress_list_posts", True), # Unified tool with prefix
("unknown", True), # Fallback to unified
]
for tool_name, expected_skip in test_cases:
# Determine is_system_tool
is_system_tool = any(tool_name.endswith(st) for st in SYSTEM_TOOLS)
# Determine is_unified_tool
if tool_name == "unknown":
is_unified_tool = True
else:
is_unified_tool = (
tool_name.startswith("wordpress_")
or tool_name.startswith("woocommerce_")
or tool_name.startswith("mcp__coolify-projects__wordpress_")
or tool_name.startswith("mcp__coolify-projects__woocommerce_")
)
# Calculate skip_project_check
skip_project_check = is_unified_tool or is_system_tool
assert skip_project_check == expected_skip, (
f"Failed for {tool_name}: expected skip={expected_skip}, got {skip_project_check} "
f"(unified={is_unified_tool}, system={is_system_tool})"
)
print(
f"{tool_name}: skip_project_check={skip_project_check} (unified={is_unified_tool}, system={is_system_tool})"
)
if __name__ == "__main__":
print("\n=== Testing System Tool Detection ===")
test_system_tool_detection()
print("\n=== Testing Unified Tool Detection ===")
test_unified_tool_detection()
print("\n=== Testing skip_project_check Logic ===")
test_skip_project_check_logic()
print("\n✅ All tests passed!")
print("\nExpected behavior after fix:")
print("1. Site-level key + list_projects → Validation passes, then gets proper error:")
print(" 'System tools require global API key (project_id=\"*\")'")
print(
"2. Site-level key + wordpress_list_posts(site=other) → Validation passes, handler blocks with:"
)
print(" 'Access denied. This API key is restricted to project X'")
print("3. Global key + any tool → Works")

View File

@@ -0,0 +1,161 @@
#!/usr/bin/env python3
"""
Quick test to verify per-project API key isolation.
This test checks that:
1. Per-project API key can access its own project
2. Per-project API key CANNOT access other projects
3. Global API key can access all projects
"""
import asyncio
import os
# Mock setup
os.environ["MASTER_API_KEY"] = "test_master_key_123"
os.environ["WORDPRESS_SITE1_URL"] = "https://site1.example.com"
os.environ["WORDPRESS_SITE1_USERNAME"] = "admin"
os.environ["WORDPRESS_SITE1_APP_PASSWORD"] = "password1"
os.environ["WORDPRESS_SITE4_URL"] = "https://site4.example.com"
os.environ["WORDPRESS_SITE4_USERNAME"] = "admin"
os.environ["WORDPRESS_SITE4_APP_PASSWORD"] = "password4"
# Import after env setup
from core.api_keys import get_api_key_manager
from core.project_manager import get_project_manager
from core.site_registry import get_site_registry
from core.unified_tools import UnifiedToolGenerator
print("=" * 60)
print("Testing Per-Project API Key Isolation")
print("=" * 60)
# Initialize
api_key_manager = get_api_key_manager()
project_manager = get_project_manager()
site_registry = get_site_registry()
# Discover sites
from plugins import registry as plugin_registry
plugin_types = plugin_registry.get_registered_types()
site_registry.discover_sites(plugin_types)
print(f"\nDiscovered sites: {list(site_registry.sites.keys())}")
# Create API keys
print("\n1. Creating API keys...")
# Global key
global_key = api_key_manager.create_key(
project_id="*", scope="admin", description="Global test key"
)
print(f" ✓ Global key created: {global_key}")
# Per-project key for wordpress_site4
site4_key = api_key_manager.create_key(
project_id="wordpress_site4", scope="admin", description="Site4 only key"
)
print(f" ✓ Site4 key created: {site4_key}")
# Create unified tool generator
unified_gen = UnifiedToolGenerator(project_manager)
unified_tools = unified_gen.generate_all_unified_tools()
print(f"\n2. Generated {len(unified_tools)} unified tools")
# Find wordpress_list_posts tool
list_posts_tool = None
for tool in unified_tools:
if tool["name"] == "wordpress_list_posts":
list_posts_tool = tool
break
if not list_posts_tool:
print(" ✗ wordpress_list_posts tool not found!")
exit(1)
print(" ✓ Found wordpress_list_posts tool")
async def test_access(key_token, site_id, expected_result):
"""Test if a key can access a site"""
from server import _api_key_context
# Validate key and set context (simulating middleware)
key_id = api_key_manager.validate_key(
key_token, project_id="*", required_scope="read", skip_project_check=True
)
if key_id:
key = api_key_manager.keys.get(key_id)
_api_key_context.set(
{
"key_id": key_id,
"project_id": key.project_id,
"scope": key.scope,
"is_global": key.project_id == "*",
}
)
# Try to call the handler
handler = list_posts_tool["handler"]
result = await handler(site=site_id, per_page=1)
# Check result
is_error = isinstance(result, str) and result.startswith("Error: Access denied")
if expected_result == "allowed":
if not is_error:
print(" ✓ Access allowed as expected")
return True
else:
print(" ✗ FAIL: Access denied but should be allowed!")
print(f" Result: {result[:100]}")
return False
else: # expected_result == "denied"
if is_error:
print(" ✓ Access denied as expected")
return True
else:
print(" ✗ FAIL: Access allowed but should be denied!")
print(f" Result: {result[:100] if isinstance(result, str) else str(result)[:100]}")
return False
async def run_tests():
"""Run all test cases"""
print("\n3. Testing access control...")
all_pass = True
# Test 1: Per-project key accessing its own project
print("\n Test 1: Per-project key (site4) → site4")
all_pass &= await test_access(site4_key, "site4", "allowed")
# Test 2: Per-project key accessing different project
print("\n Test 2: Per-project key (site4) → site1")
all_pass &= await test_access(site4_key, "site1", "denied")
# Test 3: Global key accessing any project
print("\n Test 3: Global key → site1")
all_pass &= await test_access(global_key, "site1", "allowed")
print("\n Test 4: Global key → site4")
all_pass &= await test_access(global_key, "site4", "allowed")
return all_pass
# Run tests
result = asyncio.run(run_tests())
print("\n" + "=" * 60)
if result:
print("✅ All tests PASSED!")
print("Per-project API key isolation is working correctly!")
else:
print("❌ Some tests FAILED!")
print("Per-project API key isolation has issues!")
print("=" * 60)
exit(0 if result else 1)

View File

@@ -0,0 +1,60 @@
#!/usr/bin/env python3
"""
Test context-based API key isolation after fixing circular import.
"""
from core.context import clear_api_key_context, get_api_key_context, set_api_key_context
print("=" * 60)
print("Testing Context API")
print("=" * 60)
# Test 1: Set and get context
print("\nTest 1: Set and get context")
set_api_key_context(
key_id="key_test123", project_id="wordpress_site4", scope="admin", is_global=False
)
ctx = get_api_key_context()
assert ctx["key_id"] == "key_test123"
assert ctx["project_id"] == "wordpress_site4"
assert not ctx["is_global"]
print(" ✓ PASS: Context set and retrieved correctly")
# Test 2: Check access logic
print("\nTest 2: Access logic for per-project key")
allowed_project = ctx["project_id"]
target_project = "wordpress_site4"
if allowed_project == target_project:
print(" ✓ PASS: Access allowed (same project)")
else:
print(" ✗ FAIL: Should allow access")
# Test 3: Different project access
print("\nTest 3: Access logic for different project")
target_project = "wordpress_site1"
if allowed_project != target_project:
print(" ✓ PASS: Access denied (different project)")
else:
print(" ✗ FAIL: Should deny access")
# Test 4: Global key
print("\nTest 4: Global key access")
set_api_key_context(key_id="key_global", project_id="*", scope="admin", is_global=True)
ctx = get_api_key_context()
if ctx["is_global"]:
print(" ✓ PASS: Global key bypasses project check")
else:
print(" ✗ FAIL: Should be global")
# Test 5: Clear context
print("\nTest 5: Clear context")
clear_api_key_context()
ctx = get_api_key_context()
if ctx is None:
print(" ✓ PASS: Context cleared")
else:
print(" ✗ FAIL: Context should be None")
print("\n" + "=" * 60)
print("✅ All context tests passed!")
print("=" * 60)

View File

@@ -0,0 +1,200 @@
#!/usr/bin/env python3
"""
Test script for Phase 7.2 Enhanced Health Monitoring
Tests:
1. Health monitor initialization
2. Metrics recording
3. Health checks
4. System metrics
5. Alert thresholds
6. Metrics export
"""
import asyncio
import json
import os
import sys
# Fix Windows console encoding
if sys.platform == "win32":
sys.stdout.reconfigure(encoding="utf-8")
sys.stderr.reconfigure(encoding="utf-8")
# Add project root to path
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from core import AuditLogger, ProjectManager, initialize_health_monitor
async def test_health_monitor():
"""Test health monitoring system."""
print("=" * 60)
print("Phase 7.2 - Enhanced Health Monitoring Tests")
print("=" * 60)
# Initialize components
print("\n1. Initializing components...")
project_manager = ProjectManager()
audit_logger = AuditLogger()
health_monitor = initialize_health_monitor(
project_manager=project_manager,
audit_logger=audit_logger,
metrics_retention_hours=24,
max_metrics_per_project=1000,
)
print("✅ Health monitor initialized")
print(" Retention: 24 hours")
print(" Max metrics per project: 1000")
# Test 2: Record some sample metrics
print("\n2. Recording sample metrics...")
# Simulate successful requests
for i in range(5):
health_monitor.record_request(
project_id="wordpress_site1", response_time_ms=100.0 + (i * 10), success=True
)
print("✅ Recorded 5 successful requests for wordpress_site1")
# Simulate failed requests
for i in range(2):
health_monitor.record_request(
project_id="wordpress_site1",
response_time_ms=500.0,
success=False,
error_message="Connection timeout",
)
print("✅ Recorded 2 failed requests for wordpress_site1")
# Record for another project
health_monitor.record_request(project_id="wordpress_site2", response_time_ms=80.0, success=True)
print("✅ Recorded 1 successful request for wordpress_site2")
# Test 3: Get project metrics
print("\n3. Getting project metrics...")
metrics = health_monitor.get_project_metrics("wordpress_site1", hours=1)
print("✅ Metrics for wordpress_site1:")
print(f" Total requests: {metrics['total_requests']}")
print(f" Successful: {metrics['successful_requests']}")
print(f" Failed: {metrics['failed_requests']}")
print(f" Error rate: {metrics['error_rate_percent']}%")
print(f" Avg response time: {metrics['response_time']['average_ms']}ms")
# Test 4: Get system metrics
print("\n4. Getting system metrics...")
system_metrics = health_monitor.get_system_metrics()
print("✅ System metrics:")
print(f" Uptime: {system_metrics.uptime_seconds:.2f}s")
print(f" Total requests: {system_metrics.total_requests}")
print(f" Successful: {system_metrics.successful_requests}")
print(f" Failed: {system_metrics.failed_requests}")
print(f" Error rate: {system_metrics.error_rate_percent}%")
print(f" Avg response time: {system_metrics.average_response_time_ms}ms")
# Test 5: Get uptime
print("\n5. Getting uptime...")
uptime = health_monitor.get_uptime()
print(f"✅ Uptime: {uptime['uptime_formatted']}")
# Test 6: Test alert thresholds
print("\n6. Testing alert thresholds...")
# Record a slow request to trigger alert
health_monitor.record_request(
project_id="wordpress_site3", response_time_ms=6000.0, success=True # > 5000ms threshold
)
print("✅ Recorded slow request (6000ms)")
# Record many failures to trigger error rate alert
for i in range(10):
health_monitor.record_request(
project_id="wordpress_site3",
response_time_ms=200.0,
success=False,
error_message="API error",
)
print("✅ Recorded 10 failures to trigger error rate alert")
# Test 7: Check if project exists in manager
print("\n7. Checking project health (simulation)...")
# Since we don't have actual WordPress sites running,
# we'll just show the metrics we collected
site3_metrics = health_monitor.get_project_metrics("wordpress_site3", hours=1)
print("✅ wordpress_site3 metrics:")
print(f" Total requests: {site3_metrics['total_requests']}")
print(f" Error rate: {site3_metrics['error_rate_percent']}%")
print(f" Max response time: {site3_metrics['response_time']['max_ms']}ms")
# Check for alerts
alert_data = {
"response_time_ms": site3_metrics["response_time"]["max_ms"],
"error_rate_percent": site3_metrics["error_rate_percent"],
}
alerts = health_monitor._check_alerts("wordpress_site3", alert_data)
if alerts:
print("⚠️ Alerts triggered:")
for alert in alerts:
print(f" {alert}")
else:
print("✅ No alerts")
# Test 8: Export metrics
print("\n8. Exporting metrics...")
export_path = "logs/test_metrics_export.json"
exported_file = health_monitor.export_metrics(output_path=export_path)
print(f"✅ Metrics exported to: {exported_file}")
# Verify export file
if os.path.exists(export_path):
with open(export_path, encoding="utf-8") as f:
export_data = json.load(f)
print(" Export contains:")
print(" - System metrics: ✅")
print(" - Uptime info: ✅")
print(f" - {len(export_data['projects'])} projects")
# Test 9: Custom alert threshold
print("\n9. Testing custom alert thresholds...")
health_monitor.add_alert_threshold(
project_id="wordpress_site1",
name="Custom Response Time",
metric="response_time_ms",
threshold=150.0,
comparison="gt",
severity="warning",
)
print("✅ Added custom alert threshold for wordpress_site1")
# Summary
print("\n" + "=" * 60)
print("TEST SUMMARY")
print("=" * 60)
print("✅ Health monitor initialization - PASS")
print("✅ Metrics recording - PASS")
print("✅ Project metrics retrieval - PASS")
print("✅ System metrics retrieval - PASS")
print("✅ Uptime tracking - PASS")
print("✅ Alert threshold checking - PASS")
print("✅ Metrics export - PASS")
print("✅ Custom alert thresholds - PASS")
print("\n🎉 All tests passed!")
print("=" * 60)
return True
if __name__ == "__main__":
try:
asyncio.run(test_health_monitor())
sys.exit(0)
except Exception as e:
print(f"\n❌ Test failed with error: {e}")
import traceback
traceback.print_exc()
sys.exit(1)

View File

@@ -0,0 +1,71 @@
#!/usr/bin/env python3
"""
Test OAuth metadata endpoints locally
"""
import requests
BASE_URL = "http://localhost:8000"
print("Testing OAuth Metadata Endpoints")
print("=" * 60)
# Test 1: OAuth Authorization Server Metadata
print("\n1. Testing /.well-known/oauth-authorization-server")
try:
response = requests.get(f"{BASE_URL}/.well-known/oauth-authorization-server")
print(f" Status: {response.status_code}")
if response.status_code == 200:
data = response.json()
print(f" ✓ Issuer: {data.get('issuer')}")
print(f" ✓ Authorization Endpoint: {data.get('authorization_endpoint')}")
print(f" ✓ Token Endpoint: {data.get('token_endpoint')}")
print(f" ✓ Registration Endpoint: {data.get('registration_endpoint')}")
if data.get("registration_endpoint"):
print(" ✅ RFC 7591 Dynamic Client Registration SUPPORTED")
else:
print(" ❌ RFC 7591 NOT FOUND in metadata")
else:
print(f" ❌ Failed: {response.text}")
except Exception as e:
print(f" ❌ Error: {e}")
# Test 2: OAuth Protected Resource Metadata
print("\n2. Testing /.well-known/oauth-protected-resource")
try:
response = requests.get(f"{BASE_URL}/.well-known/oauth-protected-resource")
print(f" Status: {response.status_code}")
if response.status_code == 200:
data = response.json()
print(f" ✓ Resource: {data.get('resource')}")
print(f" ✓ Scopes: {data.get('scopes_supported')}")
print(" ✅ Protected Resource metadata available")
else:
print(f" ❌ Failed: {response.text}")
except Exception as e:
print(f" ❌ Error: {e}")
# Test 3: Registration Endpoint
print("\n3. Testing POST /oauth/register")
try:
payload = {
"client_name": "Test Client",
"redirect_uris": ["http://localhost:3000/callback"],
"grant_types": ["authorization_code", "refresh_token"],
"scope": "read write",
}
response = requests.post(f"{BASE_URL}/oauth/register", json=payload)
print(f" Status: {response.status_code}")
if response.status_code == 201:
data = response.json()
print(f" ✓ Client ID: {data.get('client_id')}")
print(f" ✓ Client Secret: {data.get('client_secret')[:20]}...")
print(" ✅ Dynamic Client Registration WORKING")
else:
print(f" ❌ Failed: {response.text}")
except Exception as e:
print(f" ❌ Error: {e}")
print("\n" + "=" * 60)
print("Test Complete!")

View File

@@ -0,0 +1,124 @@
#!/usr/bin/env python3
"""
Test OAuth client registration endpoint security.
Verifies that:
1. Registration endpoint requires Master API Key
2. Unauthorized requests are rejected with 401
3. Valid Master API Key allows registration
"""
import os
import requests
BASE_URL = os.getenv("BASE_URL", "http://localhost:8000")
MASTER_API_KEY = os.getenv("MASTER_API_KEY", "your_master_key_here")
print("Testing OAuth Client Registration Security")
print("=" * 60)
# Test 1: Registration without Authorization header
print("\n1. Testing registration WITHOUT Authorization header")
try:
payload = {
"client_name": "Unauthorized Test Client",
"redirect_uris": ["http://localhost:3000/callback"],
"grant_types": ["authorization_code"],
"scope": "read",
}
response = requests.post(f"{BASE_URL}/oauth/register", json=payload)
print(f" Status: {response.status_code}")
if response.status_code == 401:
data = response.json()
print(f" ✅ Correctly rejected: {data.get('error')}")
print(f" Message: {data.get('error_description')}")
else:
print(f" ❌ SECURITY ISSUE: Should return 401, got {response.status_code}")
print(f" Response: {response.text}")
except Exception as e:
print(f" ❌ Error: {e}")
# Test 2: Registration with invalid API key
print("\n2. Testing registration WITH invalid API key")
try:
payload = {
"client_name": "Unauthorized Test Client 2",
"redirect_uris": ["http://localhost:3000/callback"],
"grant_types": ["authorization_code"],
"scope": "read",
}
headers = {"Authorization": "Bearer invalid_api_key_12345"}
response = requests.post(f"{BASE_URL}/oauth/register", json=payload, headers=headers)
print(f" Status: {response.status_code}")
if response.status_code == 401:
data = response.json()
print(f" ✅ Correctly rejected: {data.get('error')}")
print(f" Message: {data.get('error_description')}")
else:
print(f" ❌ SECURITY ISSUE: Should return 401, got {response.status_code}")
print(f" Response: {response.text}")
except Exception as e:
print(f" ❌ Error: {e}")
# Test 3: Registration with valid Master API Key
print("\n3. Testing registration WITH valid Master API Key")
try:
payload = {
"client_name": "Authorized Test Client",
"redirect_uris": ["http://localhost:3000/callback"],
"grant_types": ["authorization_code", "refresh_token"],
"scope": "read write",
}
headers = {"Authorization": f"Bearer {MASTER_API_KEY}"}
response = requests.post(f"{BASE_URL}/oauth/register", json=payload, headers=headers)
print(f" Status: {response.status_code}")
if response.status_code == 201:
data = response.json()
print(" ✅ Successfully registered client")
print(f" Client ID: {data.get('client_id')}")
print(f" Client Secret: {data.get('client_secret', '')[:20]}...")
print(f" Client Name: {data.get('client_name')}")
else:
print(f" ❌ Registration failed: {response.status_code}")
print(f" Response: {response.text}")
except Exception as e:
print(f" ❌ Error: {e}")
# Test 4: Attempt registration with wrong Authorization format
print("\n4. Testing registration with wrong Authorization format")
try:
payload = {
"client_name": "Test Client Wrong Format",
"redirect_uris": ["http://localhost:3000/callback"],
"grant_types": ["authorization_code"],
"scope": "read",
}
headers = {"Authorization": MASTER_API_KEY} # Missing "Bearer " prefix
response = requests.post(f"{BASE_URL}/oauth/register", json=payload, headers=headers)
print(f" Status: {response.status_code}")
if response.status_code == 401:
data = response.json()
print(f" ✅ Correctly rejected: {data.get('error')}")
print(f" Message: {data.get('error_description')}")
else:
print(f" ❌ Should return 401, got {response.status_code}")
except Exception as e:
print(f" ❌ Error: {e}")
print("\n" + "=" * 60)
print("Security Test Complete!")
print("\nExpected Results:")
print(" ✅ Test 1 & 2 & 4: Should return 401 Unauthorized")
print(" ✅ Test 3: Should return 201 Created with client credentials")

View File

@@ -0,0 +1,506 @@
"""
Test suite for Rate Limiter (Phase 7.3)
Tests the Token Bucket algorithm, rate limiting logic,
and middleware integration.
Run with: python test_rate_limiter.py
"""
import time
import unittest
from unittest.mock import patch
from core.rate_limiter import (
ClientRateLimitState,
RateLimitConfig,
RateLimiter,
TokenBucket,
get_rate_limiter,
)
class TestTokenBucket(unittest.TestCase):
"""Test Token Bucket algorithm implementation."""
def test_initialization(self):
"""Test bucket initializes with full capacity."""
bucket = TokenBucket(capacity=10, refill_rate=1.0)
self.assertEqual(bucket.capacity, 10)
self.assertEqual(bucket.refill_rate, 1.0)
self.assertEqual(bucket.tokens, 10.0)
def test_consume_tokens(self):
"""Test consuming tokens from bucket."""
bucket = TokenBucket(capacity=10, refill_rate=1.0)
# Should successfully consume 5 tokens
self.assertTrue(bucket.consume(5))
self.assertEqual(bucket.get_available_tokens(), 5)
# Should successfully consume 5 more tokens
self.assertTrue(bucket.consume(5))
self.assertEqual(bucket.get_available_tokens(), 0)
# Should fail to consume when empty
self.assertFalse(bucket.consume(1))
def test_token_refill(self):
"""Test tokens refill over time."""
bucket = TokenBucket(capacity=10, refill_rate=10.0) # 10 tokens/second
# Consume all tokens
bucket.consume(10)
self.assertEqual(bucket.get_available_tokens(), 0)
# Wait 0.5 seconds, should refill 5 tokens
time.sleep(0.5)
available = bucket.get_available_tokens()
# Allow some tolerance for timing variations
self.assertGreaterEqual(available, 4)
self.assertLessEqual(available, 6)
def test_refill_cap(self):
"""Test refill doesn't exceed capacity."""
bucket = TokenBucket(capacity=10, refill_rate=10.0)
# Wait long enough to refill many times over
time.sleep(2.0)
# Should be capped at capacity
self.assertEqual(bucket.get_available_tokens(), 10)
def test_wait_time_calculation(self):
"""Test wait time calculation when tokens unavailable."""
bucket = TokenBucket(capacity=10, refill_rate=2.0) # 2 tokens/second
# Consume all tokens
bucket.consume(10)
# Need 1 token, should take 0.5 seconds
wait_time = bucket.get_wait_time(1)
self.assertAlmostEqual(wait_time, 0.5, delta=0.1)
# Need 10 tokens, should take 5 seconds
wait_time = bucket.get_wait_time(10)
self.assertAlmostEqual(wait_time, 5.0, delta=0.1)
class TestRateLimitConfig(unittest.TestCase):
"""Test rate limit configuration."""
def test_default_values(self):
"""Test default configuration values."""
config = RateLimitConfig()
self.assertEqual(config.per_minute, 60)
self.assertEqual(config.per_hour, 1000)
self.assertEqual(config.per_day, 10000)
def test_custom_values(self):
"""Test custom configuration values."""
config = RateLimitConfig(per_minute=100, per_hour=2000, per_day=20000)
self.assertEqual(config.per_minute, 100)
self.assertEqual(config.per_hour, 2000)
self.assertEqual(config.per_day, 20000)
def test_from_env(self):
"""Test loading configuration from environment."""
with patch.dict(
"os.environ",
{
"RATE_LIMIT_PER_MINUTE": "120",
"RATE_LIMIT_PER_HOUR": "2000",
"RATE_LIMIT_PER_DAY": "15000",
},
):
config = RateLimitConfig.from_env()
self.assertEqual(config.per_minute, 120)
self.assertEqual(config.per_hour, 2000)
self.assertEqual(config.per_day, 15000)
def test_from_env_with_prefix(self):
"""Test loading configuration with prefix."""
with patch.dict(
"os.environ",
{
"WORDPRESS_RATE_LIMIT_PER_MINUTE": "80",
"WORDPRESS_RATE_LIMIT_PER_HOUR": "1500",
"WORDPRESS_RATE_LIMIT_PER_DAY": "12000",
},
):
config = RateLimitConfig.from_env(prefix="WORDPRESS")
self.assertEqual(config.per_minute, 80)
self.assertEqual(config.per_hour, 1500)
self.assertEqual(config.per_day, 12000)
class TestClientRateLimitState(unittest.TestCase):
"""Test client rate limit state tracking."""
def test_initialization(self):
"""Test client state initialization."""
state = ClientRateLimitState(
client_id="test_client",
minute_bucket=TokenBucket(60, 1.0),
hour_bucket=TokenBucket(1000, 1.0),
day_bucket=TokenBucket(10000, 1.0),
)
self.assertEqual(state.client_id, "test_client")
self.assertEqual(state.total_requests, 0)
self.assertEqual(state.rejected_requests, 0)
def test_successful_request(self):
"""Test successful request consumes tokens."""
state = ClientRateLimitState(
client_id="test_client",
minute_bucket=TokenBucket(60, 1.0),
hour_bucket=TokenBucket(1000, 1.0),
day_bucket=TokenBucket(10000, 1.0),
)
allowed, message, retry_after = state.check_and_consume()
self.assertTrue(allowed)
self.assertEqual(message, "")
self.assertEqual(retry_after, 0.0)
self.assertEqual(state.total_requests, 1)
self.assertEqual(state.rejected_requests, 0)
def test_minute_limit_exceeded(self):
"""Test rejection when per-minute limit exceeded."""
state = ClientRateLimitState(
client_id="test_client",
minute_bucket=TokenBucket(2, 1.0), # Only 2 requests/minute
hour_bucket=TokenBucket(1000, 1.0),
day_bucket=TokenBucket(10000, 1.0),
)
# First 2 requests should succeed
state.check_and_consume()
state.check_and_consume()
# Third request should fail
allowed, message, retry_after = state.check_and_consume()
self.assertFalse(allowed)
self.assertIn("per minute", message.lower())
self.assertGreater(retry_after, 0)
self.assertEqual(state.rejected_requests, 1)
def test_hour_limit_exceeded(self):
"""Test rejection when per-hour limit exceeded."""
state = ClientRateLimitState(
client_id="test_client",
minute_bucket=TokenBucket(100, 100.0), # High minute limit
hour_bucket=TokenBucket(2, 1.0), # Only 2 requests/hour
day_bucket=TokenBucket(10000, 1.0),
)
# First 2 requests should succeed
state.check_and_consume()
state.check_and_consume()
# Third request should fail at hour limit
allowed, message, retry_after = state.check_and_consume()
self.assertFalse(allowed)
self.assertIn("per hour", message.lower())
def test_day_limit_exceeded(self):
"""Test rejection when daily limit exceeded."""
state = ClientRateLimitState(
client_id="test_client",
minute_bucket=TokenBucket(100, 100.0),
hour_bucket=TokenBucket(100, 100.0),
day_bucket=TokenBucket(2, 1.0), # Only 2 requests/day
)
# First 2 requests should succeed
state.check_and_consume()
state.check_and_consume()
# Third request should fail at day limit
allowed, message, retry_after = state.check_and_consume()
self.assertFalse(allowed)
self.assertIn("daily", message.lower())
def test_token_refund_on_rejection(self):
"""Test tokens are refunded when request rejected at later stage."""
state = ClientRateLimitState(
client_id="test_client",
minute_bucket=TokenBucket(10, 1.0),
hour_bucket=TokenBucket(2, 1.0), # Will fail here
day_bucket=TokenBucket(10, 1.0),
)
# Consume hour bucket
state.check_and_consume()
state.check_and_consume()
# Check minute bucket before rejection
minute_before = state.minute_bucket.get_available_tokens()
# This should fail at hour limit
allowed, message, retry_after = state.check_and_consume()
self.assertFalse(allowed)
# Minute bucket should be refunded
minute_after = state.minute_bucket.get_available_tokens()
self.assertEqual(minute_before, minute_after)
def test_get_stats(self):
"""Test statistics generation."""
state = ClientRateLimitState(
client_id="test_client",
minute_bucket=TokenBucket(60, 1.0),
hour_bucket=TokenBucket(1000, 1.0),
day_bucket=TokenBucket(10000, 1.0),
)
# Make some requests
state.check_and_consume()
state.check_and_consume()
stats = state.get_stats()
self.assertEqual(stats["client_id"], "test_client")
self.assertEqual(stats["total_requests"], 2)
self.assertEqual(stats["rejected_requests"], 0)
self.assertEqual(stats["success_rate"], 1.0)
self.assertIn("available_tokens", stats)
self.assertIn("limits", stats)
class TestRateLimiter(unittest.TestCase):
"""Test RateLimiter class."""
def setUp(self):
"""Create a fresh rate limiter for each test."""
# Reset singleton
import core.rate_limiter
core.rate_limiter._rate_limiter = None
self.limiter = RateLimiter()
def test_initialization(self):
"""Test rate limiter initialization."""
self.assertIsNotNone(self.limiter.default_config)
self.assertEqual(len(self.limiter.clients), 0)
self.assertEqual(self.limiter.global_stats["total_requests"], 0)
def test_singleton_pattern(self):
"""Test get_rate_limiter returns singleton."""
limiter1 = get_rate_limiter()
limiter2 = get_rate_limiter()
self.assertIs(limiter1, limiter2)
def test_client_creation(self):
"""Test automatic client state creation."""
allowed, message, retry_after = self.limiter.check_rate_limit(
client_id="client1", tool_name="test_tool"
)
self.assertTrue(allowed)
self.assertIn("client1", self.limiter.clients)
def test_successful_rate_limit_check(self):
"""Test successful rate limit check."""
allowed, message, retry_after = self.limiter.check_rate_limit(
client_id="client1", tool_name="test_tool"
)
self.assertTrue(allowed)
self.assertEqual(message, "")
self.assertEqual(retry_after, 0.0)
self.assertEqual(self.limiter.global_stats["total_requests"], 1)
self.assertEqual(self.limiter.global_stats["total_rejected"], 0)
def test_rate_limit_exceeded(self):
"""Test rate limit exceeded scenario."""
# Configure very low limits
self.limiter.configure_limits("test_plugin", per_minute=2, per_hour=2, per_day=2)
# First 2 requests should succeed
self.limiter.check_rate_limit("client1", "test", "test_plugin")
self.limiter.check_rate_limit("client1", "test", "test_plugin")
# Third should fail
allowed, message, retry_after = self.limiter.check_rate_limit(
"client1", "test", "test_plugin"
)
self.assertFalse(allowed)
self.assertGreater(retry_after, 0)
self.assertEqual(self.limiter.global_stats["total_rejected"], 1)
def test_plugin_specific_limits(self):
"""Test plugin-specific rate limits."""
# Configure different limits for wordpress
self.limiter.configure_limits("wordpress", per_minute=5, per_hour=5, per_day=5)
# Make 5 wordpress requests
for _i in range(5):
allowed, _, _ = self.limiter.check_rate_limit("client1", "wordpress_tool", "wordpress")
self.assertTrue(allowed)
# 6th should fail
allowed, message, _ = self.limiter.check_rate_limit(
"client1", "wordpress_tool", "wordpress"
)
self.assertFalse(allowed)
def test_multiple_clients(self):
"""Test multiple clients tracked separately."""
# Configure low limits
self.limiter.configure_limits("test", per_minute=2, per_hour=2, per_day=2)
# Client 1 makes 2 requests
self.limiter.check_rate_limit("client1", "test", "test")
self.limiter.check_rate_limit("client1", "test", "test")
# Client 1's third request should fail
allowed, _, _ = self.limiter.check_rate_limit("client1", "test", "test")
self.assertFalse(allowed)
# But client 2's first request should succeed
allowed, _, _ = self.limiter.check_rate_limit("client2", "test", "test")
self.assertTrue(allowed)
def test_get_client_stats(self):
"""Test getting client statistics."""
# Make some requests
self.limiter.check_rate_limit("client1", "test")
self.limiter.check_rate_limit("client1", "test")
stats = self.limiter.get_client_stats("client1")
self.assertIsNotNone(stats)
self.assertEqual(stats["client_id"], "client1")
self.assertEqual(stats["total_requests"], 2)
def test_get_client_stats_not_found(self):
"""Test getting stats for non-existent client."""
stats = self.limiter.get_client_stats("nonexistent")
self.assertIsNone(stats)
def test_get_all_stats(self):
"""Test getting all statistics."""
# Make requests from multiple clients
self.limiter.check_rate_limit("client1", "test")
self.limiter.check_rate_limit("client2", "test")
stats = self.limiter.get_all_stats()
self.assertIn("global", stats)
self.assertIn("default_limits", stats)
self.assertIn("plugin_limits", stats)
self.assertIn("clients", stats)
self.assertEqual(stats["global"]["total_requests"], 2)
self.assertEqual(len(stats["clients"]), 2)
def test_reset_client(self):
"""Test resetting specific client."""
# Make some requests
self.limiter.check_rate_limit("client1", "test")
self.assertIn("client1", self.limiter.clients)
# Reset client
success = self.limiter.reset_client("client1")
self.assertTrue(success)
self.assertNotIn("client1", self.limiter.clients)
def test_reset_client_not_found(self):
"""Test resetting non-existent client."""
success = self.limiter.reset_client("nonexistent")
self.assertFalse(success)
def test_reset_all(self):
"""Test resetting all clients."""
# Make requests from multiple clients
self.limiter.check_rate_limit("client1", "test")
self.limiter.check_rate_limit("client2", "test")
self.limiter.check_rate_limit("client3", "test")
# Reset all
count = self.limiter.reset_all()
self.assertEqual(count, 3)
self.assertEqual(len(self.limiter.clients), 0)
self.assertEqual(self.limiter.global_stats["total_requests"], 0)
def test_configure_limits(self):
"""Test dynamic limit configuration."""
self.limiter.configure_limits("custom_plugin", per_minute=100, per_hour=2000, per_day=20000)
config = self.limiter.plugin_configs["custom_plugin"]
self.assertEqual(config.per_minute, 100)
self.assertEqual(config.per_hour, 2000)
self.assertEqual(config.per_day, 20000)
class TestIntegration(unittest.TestCase):
"""Integration tests simulating real-world usage."""
def setUp(self):
"""Create fresh rate limiter."""
import core.rate_limiter
core.rate_limiter._rate_limiter = None
self.limiter = RateLimiter()
def test_burst_traffic(self):
"""Test handling of burst traffic within limits."""
# Configure to allow 10/minute
self.limiter.configure_limits("test", per_minute=10, per_hour=100, per_day=1000)
# Simulate burst of 10 requests
success_count = 0
for _i in range(10):
allowed, _, _ = self.limiter.check_rate_limit("client1", "test", "test")
if allowed:
success_count += 1
# All 10 should succeed (burst allowed)
self.assertEqual(success_count, 10)
# 11th should fail
allowed, _, _ = self.limiter.check_rate_limit("client1", "test", "test")
self.assertFalse(allowed)
def test_sustained_traffic(self):
"""Test handling of sustained traffic with refill."""
# Configure high refill rate for testing
self.limiter.configure_limits("test", per_minute=5, per_hour=100, per_day=1000)
# Make 5 requests (consume all minute tokens)
for _i in range(5):
self.limiter.check_rate_limit("client1", "test", "test")
# Should be rate limited now
allowed, _, _ = self.limiter.check_rate_limit("client1", "test", "test")
self.assertFalse(allowed)
# Wait for refill (should get ~2-3 tokens in 0.5 seconds at 5/min rate)
time.sleep(0.5)
# Should be able to make 1-2 more requests
allowed, _, _ = self.limiter.check_rate_limit("client1", "test", "test")
# May or may not succeed depending on exact timing
# Just verify no crash and reasonable behavior
def run_tests():
"""Run all tests."""
# Create test suite
loader = unittest.TestLoader()
suite = unittest.TestSuite()
# Add all test classes
suite.addTests(loader.loadTestsFromTestCase(TestTokenBucket))
suite.addTests(loader.loadTestsFromTestCase(TestRateLimitConfig))
suite.addTests(loader.loadTestsFromTestCase(TestClientRateLimitState))
suite.addTests(loader.loadTestsFromTestCase(TestRateLimiter))
suite.addTests(loader.loadTestsFromTestCase(TestIntegration))
# Run tests
runner = unittest.TextTestRunner(verbosity=2)
result = runner.run(suite)
# Return exit code
return 0 if result.wasSuccessful() else 1
if __name__ == "__main__":
import sys
sys.exit(run_tests())

View File

@@ -0,0 +1,133 @@
#!/usr/bin/env python3
"""
Simple standalone test to verify the API key isolation logic.
"""
# Simulate the context and validation logic
from contextvars import ContextVar
# Create a mock context var
_api_key_context = ContextVar("api_key_context", default=None)
def test_case_1():
"""Per-project key accessing its own project"""
print("\nTest 1: Per-project key (site4) accessing site4")
# Set context for per-project key
_api_key_context.set(
{"key_id": "key_123", "project_id": "wordpress_site4", "scope": "admin", "is_global": False}
)
# Check access
api_key_info = _api_key_context.get()
full_id = "wordpress_site4" # Target project
if api_key_info and not api_key_info.get("is_global"):
allowed_project = api_key_info.get("project_id")
if allowed_project != full_id:
print(" ✗ FAIL: Access denied (should be allowed)")
return False
else:
print(" ✓ PASS: Access allowed")
return True
else:
print(" ✓ PASS: Access allowed (global key)")
return True
def test_case_2():
"""Per-project key accessing different project"""
print("\nTest 2: Per-project key (site4) accessing site1")
# Set context for per-project key
_api_key_context.set(
{"key_id": "key_123", "project_id": "wordpress_site4", "scope": "admin", "is_global": False}
)
# Check access
api_key_info = _api_key_context.get()
full_id = "wordpress_site1" # Different project
if api_key_info and not api_key_info.get("is_global"):
allowed_project = api_key_info.get("project_id")
if allowed_project != full_id:
print(" ✓ PASS: Access denied (as expected)")
return True
else:
print(" ✗ FAIL: Access allowed (should be denied)")
return False
else:
print(" ✗ FAIL: Access allowed (should be denied)")
return False
def test_case_3():
"""Global key accessing any project"""
print("\nTest 3: Global key accessing site1")
# Set context for global key
_api_key_context.set(
{"key_id": "key_456", "project_id": "*", "scope": "admin", "is_global": True}
)
# Check access
api_key_info = _api_key_context.get()
full_id = "wordpress_site1"
if api_key_info and not api_key_info.get("is_global"):
allowed_project = api_key_info.get("project_id")
if allowed_project != full_id:
print(" ✗ FAIL: Access denied (should be allowed)")
return False
else:
print(" ✓ PASS: Access allowed")
return True
else:
print(" ✓ PASS: Access allowed (global key)")
return True
def test_case_4():
"""Global key accessing different project"""
print("\nTest 4: Global key accessing site4")
# Set context for global key
_api_key_context.set(
{"key_id": "key_456", "project_id": "*", "scope": "admin", "is_global": True}
)
# Check access
api_key_info = _api_key_context.get()
full_id = "wordpress_site4"
if api_key_info and not api_key_info.get("is_global"):
allowed_project = api_key_info.get("project_id")
if allowed_project != full_id:
print(" ✗ FAIL: Access denied (should be allowed)")
return False
else:
print(" ✓ PASS: Access allowed")
return True
else:
print(" ✓ PASS: Access allowed (global key)")
return True
# Run all tests
print("=" * 60)
print("Testing API Key Isolation Logic")
print("=" * 60)
results = [test_case_1(), test_case_2(), test_case_3(), test_case_4()]
print("\n" + "=" * 60)
if all(results):
print("✅ All tests PASSED!")
print("The isolation logic is correct!")
else:
print("❌ Some tests FAILED!")
print("The isolation logic has issues!")
print("=" * 60)
exit(0 if all(results) else 1)

View File

@@ -0,0 +1,309 @@
#!/usr/bin/env python3
"""
Integration Test Script for WordPress MCP Server
Usage:
python tests/run_integration_tests.py
Requirements:
- Server must be running
- WordPress sites configured in environment variables
- Valid API keys set up
This script tests:
1. Plugin imports and initialization
2. Handler functionality with real WordPress sites
3. Error handling
4. Tool specifications
Results are output as JSON for easy parsing.
"""
import json
import os
import sys
from datetime import UTC, datetime
# Add parent directory to path
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
class IntegrationTester:
"""Run integration tests for WordPress MCP server"""
def __init__(self):
self.results = {
"timestamp": datetime.now(UTC).isoformat(),
"tests": [],
"summary": {"total": 0, "passed": 0, "failed": 0, "skipped": 0},
}
def add_result(self, test_name, status, message="", details=None):
"""Add a test result"""
self.results["tests"].append(
{"name": test_name, "status": status, "message": message, "details": details or {}}
)
self.results["summary"]["total"] += 1
self.results["summary"][status] += 1
def test_imports(self):
"""Test 1: Import all modules"""
try:
self.add_result("imports", "passed", "All modules imported successfully")
return True
except Exception as e:
self.add_result("imports", "failed", f"Import error: {str(e)}")
return False
def test_base_plugin(self):
"""Test 2: BasePlugin architecture"""
try:
from plugins.base import BasePlugin
from plugins.wordpress.plugin import WordPressPlugin
# Check that get_tools is not abstract
assert hasattr(BasePlugin, "get_tools"), "BasePlugin missing get_tools"
assert hasattr(
BasePlugin, "get_tool_specifications"
), "BasePlugin missing get_tool_specifications"
# Check WordPress plugin has get_tool_specifications
assert hasattr(
WordPressPlugin, "get_tool_specifications"
), "WordPress missing get_tool_specifications"
# Get tool specifications without instantiation
specs = WordPressPlugin.get_tool_specifications()
self.add_result(
"base_plugin",
"passed",
"BasePlugin architecture correct",
{"tool_count": len(specs)},
)
return True
except Exception as e:
self.add_result("base_plugin", "failed", f"BasePlugin error: {str(e)}")
return False
def test_wordpress_plugin_init(self):
"""Test 3: WordPress plugin initialization"""
try:
from plugins.wordpress.plugin import WordPressPlugin
# Try to create instance with dummy config
config = {"url": "https://example.com", "username": "test", "app_password": "test"}
plugin = WordPressPlugin(config)
# Check handlers initialized
assert hasattr(plugin, "posts"), "Missing posts handler"
assert hasattr(plugin, "media"), "Missing media handler"
assert hasattr(plugin, "products"), "Missing products handler"
assert hasattr(plugin, "orders"), "Missing orders handler"
self.add_result(
"wordpress_plugin_init", "passed", "WordPress plugin initializes correctly"
)
return True
except Exception as e:
self.add_result("wordpress_plugin_init", "failed", f"Plugin init error: {str(e)}")
return False
def test_tool_specifications(self):
"""Test 4: Tool specifications format"""
try:
from plugins.wordpress.plugin import WordPressPlugin
specs = WordPressPlugin.get_tool_specifications()
if not specs:
self.add_result("tool_specifications", "failed", "No tool specifications returned")
return False
# Check first spec has required fields
first_spec = specs[0]
required_fields = ["name", "method_name", "description", "schema", "scope"]
for field in required_fields:
if field not in first_spec:
self.add_result(
"tool_specifications",
"failed",
f"Tool spec missing required field: {field}",
)
return False
self.add_result(
"tool_specifications",
"passed",
f"Tool specifications format correct ({len(specs)} tools)",
{"total_tools": len(specs), "sample_tool": first_spec.get("name")},
)
return True
except Exception as e:
self.add_result("tool_specifications", "failed", f"Tool specifications error: {str(e)}")
return False
def test_handlers_structure(self):
"""Test 5: Handlers structure"""
try:
from plugins.wordpress import handlers
expected_handlers = [
"PostsHandler",
"MediaHandler",
"TaxonomyHandler",
"CommentsHandler",
"UsersHandler",
"SiteHandler",
"ProductsHandler",
"OrdersHandler",
"CustomersHandler",
"ReportsHandler",
"CouponsHandler",
"SEOHandler",
"WPCLIHandler",
"MenusHandler",
]
missing = []
for handler_name in expected_handlers:
if not hasattr(handlers, handler_name):
missing.append(handler_name)
if missing:
self.add_result(
"handlers_structure", "failed", f"Missing handlers: {', '.join(missing)}"
)
return False
self.add_result(
"handlers_structure", "passed", f"All {len(expected_handlers)} handlers present"
)
return True
except Exception as e:
self.add_result("handlers_structure", "failed", f"Handlers structure error: {str(e)}")
return False
def test_pydantic_schemas(self):
"""Test 6: Pydantic schemas"""
try:
from plugins.wordpress.schemas import (
PaginationParams,
)
# Test basic validation
pagination = PaginationParams(per_page=10, page=1)
assert pagination.per_page == 10
assert pagination.page == 1
# Test validation errors
try:
PaginationParams(per_page=200, page=1) # Should fail (max 100)
self.add_result(
"pydantic_schemas",
"failed",
"Pydantic validation not working (should reject per_page > 100)",
)
return False
except Exception:
pass # Expected error
self.add_result("pydantic_schemas", "passed", "Pydantic schemas validate correctly")
return True
except Exception as e:
self.add_result("pydantic_schemas", "failed", f"Pydantic schemas error: {str(e)}")
return False
def test_option_b_integration(self):
"""Test 7: Option B architecture integration"""
try:
from core import SiteManager, ToolGenerator
# Create site manager
site_manager = SiteManager()
# Create tool generator
tool_generator = ToolGenerator(site_manager)
# This would normally generate tools, but we can't test without real sites
# Just check the method exists
assert hasattr(tool_generator, "generate_tools")
self.add_result(
"option_b_integration", "passed", "Option B architecture integration ready"
)
return True
except Exception as e:
self.add_result(
"option_b_integration", "failed", f"Option B integration error: {str(e)}"
)
return False
def run_all_tests(self):
"""Run all tests"""
print("🧪 Running Integration Tests for WordPress MCP Server")
print("=" * 60)
tests = [
self.test_imports,
self.test_base_plugin,
self.test_wordpress_plugin_init,
self.test_tool_specifications,
self.test_handlers_structure,
self.test_pydantic_schemas,
self.test_option_b_integration,
]
for test in tests:
test_name = test.__name__.replace("test_", "").replace("_", " ").title()
print(f"\n📋 Running: {test_name}...", end=" ")
try:
success = test()
if success:
print("✅ PASSED")
else:
print("❌ FAILED")
except Exception as e:
print(f"❌ ERROR: {str(e)}")
self.add_result(test.__name__, "failed", f"Unexpected error: {str(e)}")
# Print summary
print("\n" + "=" * 60)
print("📊 Test Summary:")
print(f" Total: {self.results['summary']['total']}")
print(f" ✅ Passed: {self.results['summary']['passed']}")
print(f" ❌ Failed: {self.results['summary']['failed']}")
print(f" ⏭️ Skipped: {self.results['summary']['skipped']}")
# Print detailed results
print("\n📝 Detailed Results:")
for test in self.results["tests"]:
status_emoji = "" if test["status"] == "passed" else ""
print(f" {status_emoji} {test['name']}: {test['message']}")
return self.results
def main():
"""Main entry point"""
tester = IntegrationTester()
results = tester.run_all_tests()
# Output JSON for programmatic parsing
print("\n" + "=" * 60)
print("📄 JSON Results:")
print(json.dumps(results, indent=2))
# Exit with error code if any tests failed
if results["summary"]["failed"] > 0:
sys.exit(1)
else:
sys.exit(0)
if __name__ == "__main__":
main()

376
tests/test_api_keys.py Normal file
View File

@@ -0,0 +1,376 @@
"""
Tests for API Keys Management System
"""
import json
import tempfile
from datetime import datetime
from pathlib import Path
import pytest
from core.api_keys import APIKeyManager
@pytest.fixture
def temp_storage():
"""Create temporary storage file for testing."""
with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".json") as f:
storage_path = f.name
yield storage_path
# Cleanup
Path(storage_path).unlink(missing_ok=True)
@pytest.fixture
def manager(temp_storage):
"""Create API Key Manager instance for testing."""
return APIKeyManager(storage_path=temp_storage)
class TestAPIKeyCreation:
"""Test API key creation."""
def test_create_key_basic(self, manager):
"""Test basic key creation."""
result = manager.create_key(project_id="wordpress_site1", scope="read")
assert "key" in result
assert "key_id" in result
assert result["key"].startswith("cmp_")
assert result["scope"] == "read"
assert result["project_id"] == "wordpress_site1"
def test_create_key_with_expiration(self, manager):
"""Test key creation with expiration."""
result = manager.create_key(project_id="wordpress_site1", scope="write", expires_in_days=30)
assert result["expires_at"] is not None
expires = datetime.fromisoformat(result["expires_at"])
now = datetime.now()
assert expires > now
assert (expires - now).days <= 30
def test_create_key_with_description(self, manager):
"""Test key creation with description."""
result = manager.create_key(
project_id="wordpress_site1", scope="admin", description="Test key for CI/CD"
)
key_id = result["key_id"]
key_info = manager.get_key_info(key_id)
assert key_info["description"] == "Test key for CI/CD"
def test_create_global_key(self, manager):
"""Test creation of global key (all projects)."""
result = manager.create_key(project_id="*", scope="admin")
assert result["project_id"] == "*"
def test_multiple_keys_same_project(self, manager):
"""Test creating multiple keys for same project."""
manager.create_key("wordpress_site1", "read")
manager.create_key("wordpress_site1", "write")
manager.create_key("wordpress_site1", "admin")
keys = manager.list_keys(project_id="wordpress_site1")
assert len(keys) == 3
scopes = {k["scope"] for k in keys}
assert scopes == {"read", "write", "admin"}
class TestAPIKeyValidation:
"""Test API key validation."""
def test_validate_key_success(self, manager):
"""Test successful key validation."""
result = manager.create_key("wordpress_site1", "read")
api_key = result["key"]
key_id = manager.validate_key(api_key, project_id="wordpress_site1", required_scope="read")
assert key_id is not None
assert key_id == result["key_id"]
def test_validate_key_wrong_project(self, manager):
"""Test validation fails for wrong project."""
result = manager.create_key("wordpress_site1", "read")
api_key = result["key"]
key_id = manager.validate_key(api_key, project_id="wordpress_site2", required_scope="read")
assert key_id is None
def test_validate_key_insufficient_scope(self, manager):
"""Test validation fails with insufficient scope."""
result = manager.create_key("wordpress_site1", "read")
api_key = result["key"]
key_id = manager.validate_key(api_key, project_id="wordpress_site1", required_scope="write")
assert key_id is None
def test_validate_key_scope_hierarchy(self, manager):
"""Test scope hierarchy: admin > write > read."""
# Admin key can do write operations
admin_result = manager.create_key("wordpress_site1", "admin")
admin_key = admin_result["key"]
key_id = manager.validate_key(
admin_key, project_id="wordpress_site1", required_scope="write"
)
assert key_id is not None
# Write key can do read operations
write_result = manager.create_key("wordpress_site1", "write")
write_key = write_result["key"]
key_id = manager.validate_key(
write_key, project_id="wordpress_site1", required_scope="read"
)
assert key_id is not None
def test_validate_global_key(self, manager):
"""Test global key works for any project."""
result = manager.create_key("*", "admin")
api_key = result["key"]
# Should work for any project
key_id = manager.validate_key(api_key, project_id="wordpress_site1", required_scope="read")
assert key_id is not None
key_id = manager.validate_key(api_key, project_id="wordpress_site2", required_scope="write")
assert key_id is not None
def test_validate_invalid_key(self, manager):
"""Test validation fails for invalid key."""
key_id = manager.validate_key(
"cmp_invalid_key", project_id="wordpress_site1", required_scope="read"
)
assert key_id is None
def test_validate_updates_usage(self, manager):
"""Test validation updates usage tracking."""
result = manager.create_key("wordpress_site1", "read")
api_key = result["key"]
key_id = result["key_id"]
# Initial state
info = manager.get_key_info(key_id)
assert info["usage_count"] == 0
assert info["last_used_at"] is None
# Use the key
manager.validate_key(api_key, project_id="wordpress_site1", required_scope="read")
# Check updated
info = manager.get_key_info(key_id)
assert info["usage_count"] == 1
assert info["last_used_at"] is not None
class TestAPIKeyRevocation:
"""Test API key revocation."""
def test_revoke_key(self, manager):
"""Test key revocation."""
result = manager.create_key("wordpress_site1", "read")
api_key = result["key"]
key_id = result["key_id"]
# Key should work before revocation
validation = manager.validate_key(
api_key, project_id="wordpress_site1", required_scope="read"
)
assert validation is not None
# Revoke key
success = manager.revoke_key(key_id)
assert success is True
# Key should not work after revocation
validation = manager.validate_key(
api_key, project_id="wordpress_site1", required_scope="read"
)
assert validation is None
def test_revoke_nonexistent_key(self, manager):
"""Test revoking non-existent key."""
success = manager.revoke_key("key_nonexistent")
assert success is False
def test_delete_key(self, manager):
"""Test permanent key deletion."""
result = manager.create_key("wordpress_site1", "read")
key_id = result["key_id"]
# Delete key
success = manager.delete_key(key_id)
assert success is True
# Key should not exist
info = manager.get_key_info(key_id)
assert info is None
class TestAPIKeyExpiration:
"""Test API key expiration."""
def test_expired_key_invalid(self, manager):
"""Test expired key is invalid."""
# Create key that expires immediately
result = manager.create_key(
"wordpress_site1", "read", expires_in_days=-1 # Already expired
)
api_key = result["key"]
# Should not validate
key_id = manager.validate_key(api_key, project_id="wordpress_site1", required_scope="read")
assert key_id is None
def test_key_expiration_check(self, manager):
"""Test expiration checking."""
# Non-expiring key
result1 = manager.create_key("wordpress_site1", "read")
key1_id = result1["key_id"]
info1 = manager.get_key_info(key1_id)
assert info1["expired"] is False
# Expired key
result2 = manager.create_key("wordpress_site1", "read", expires_in_days=-1)
key2_id = result2["key_id"]
info2 = manager.get_key_info(key2_id)
assert info2["expired"] is True
class TestAPIKeyListing:
"""Test API key listing."""
def test_list_all_keys(self, manager):
"""Test listing all keys."""
manager.create_key("wordpress_site1", "read")
manager.create_key("wordpress_site2", "write")
manager.create_key("wordpress_site3", "admin")
keys = manager.list_keys()
assert len(keys) == 3
def test_list_keys_by_project(self, manager):
"""Test filtering keys by project."""
manager.create_key("wordpress_site1", "read")
manager.create_key("wordpress_site1", "write")
manager.create_key("wordpress_site2", "admin")
keys = manager.list_keys(project_id="wordpress_site1")
assert len(keys) == 2
def test_list_keys_exclude_revoked(self, manager):
"""Test excluding revoked keys from listing."""
result1 = manager.create_key("wordpress_site1", "read")
manager.create_key("wordpress_site1", "write")
# Revoke one key
manager.revoke_key(result1["key_id"])
# List without revoked
keys = manager.list_keys(include_revoked=False)
assert len(keys) == 1
# List with revoked
keys = manager.list_keys(include_revoked=True)
assert len(keys) == 2
class TestAPIKeyRotation:
"""Test API key rotation."""
def test_rotate_keys(self, manager):
"""Test key rotation for a project."""
# Create keys
manager.create_key("wordpress_site1", "read")
manager.create_key("wordpress_site1", "write")
# Rotate
new_keys = manager.rotate_keys("wordpress_site1")
assert len(new_keys) == 2
# All new keys should have different IDs
new_key_ids = {k["key_id"] for k in new_keys}
assert len(new_key_ids) == 2
# Old keys should be revoked
all_keys = manager.list_keys(project_id="wordpress_site1", include_revoked=True)
revoked_count = sum(1 for k in all_keys if k["revoked"])
assert revoked_count == 2
def test_rotate_preserves_scopes(self, manager):
"""Test rotation preserves scopes."""
manager.create_key("wordpress_site1", "read", description="Read key")
manager.create_key("wordpress_site1", "admin", description="Admin key")
new_keys = manager.rotate_keys("wordpress_site1")
scopes = {k["scope"] for k in new_keys}
assert scopes == {"read", "admin"}
class TestAPIKeyPersistence:
"""Test API key persistence."""
def test_keys_persist_across_instances(self, temp_storage):
"""Test keys are saved and loaded correctly."""
# Create manager and add key
manager1 = APIKeyManager(storage_path=temp_storage)
result = manager1.create_key("wordpress_site1", "read")
key_id = result["key_id"]
# Create new manager instance with same storage
manager2 = APIKeyManager(storage_path=temp_storage)
# Key should exist
info = manager2.get_key_info(key_id)
assert info is not None
assert info["project_id"] == "wordpress_site1"
def test_storage_file_format(self, temp_storage):
"""Test storage file is valid JSON."""
manager = APIKeyManager(storage_path=temp_storage)
manager.create_key("wordpress_site1", "read")
# Check file is valid JSON
with open(temp_storage) as f:
data = json.load(f)
assert isinstance(data, dict)
assert len(data) == 1
class TestAPIKeyInfo:
"""Test getting key information."""
def test_get_key_info(self, manager):
"""Test getting key information."""
result = manager.create_key("wordpress_site1", "admin", description="Test key")
key_id = result["key_id"]
info = manager.get_key_info(key_id)
assert info is not None
assert info["key_id"] == key_id
assert info["project_id"] == "wordpress_site1"
assert info["scope"] == "admin"
assert info["description"] == "Test key"
assert info["valid"] is True
def test_get_nonexistent_key_info(self, manager):
"""Test getting info for non-existent key."""
info = manager.get_key_info("key_nonexistent")
assert info is None
# Run tests
if __name__ == "__main__":
pytest.main([__file__, "-v"])

115
tests/test_auth.py Normal file
View File

@@ -0,0 +1,115 @@
"""Tests for Authentication System (core/auth.py)."""
import os
import secrets
import pytest
from core.auth import AuthManager
@pytest.fixture
def auth_with_key(monkeypatch):
"""Create AuthManager with a known master key."""
monkeypatch.setenv("MASTER_API_KEY", "test-master-key-12345")
return AuthManager()
@pytest.fixture
def auth_without_key(monkeypatch):
"""Create AuthManager without env key (auto-generates)."""
monkeypatch.delenv("MASTER_API_KEY", raising=False)
return AuthManager()
class TestMasterKeyInit:
"""Test master key initialization."""
def test_loads_key_from_env(self, auth_with_key):
"""Master key should be loaded from MASTER_API_KEY env var."""
assert auth_with_key.master_api_key == "test-master-key-12345"
def test_generates_key_when_missing(self, auth_without_key):
"""Should auto-generate a key when env var is not set."""
assert auth_without_key.master_api_key is not None
assert len(auth_without_key.master_api_key) > 20
def test_generated_keys_are_unique(self, monkeypatch):
"""Each init without env should produce a different key."""
monkeypatch.delenv("MASTER_API_KEY", raising=False)
a = AuthManager()
b = AuthManager()
assert a.master_api_key != b.master_api_key
class TestMasterKeyValidation:
"""Test master key validation."""
def test_valid_key_accepted(self, auth_with_key):
"""Correct master key should be accepted."""
assert auth_with_key.validate_master_key("test-master-key-12345") is True
def test_invalid_key_rejected(self, auth_with_key):
"""Wrong master key should be rejected."""
assert auth_with_key.validate_master_key("wrong-key") is False
def test_empty_key_rejected(self, auth_with_key):
"""Empty string should be rejected."""
assert auth_with_key.validate_master_key("") is False
def test_timing_safe_comparison(self, auth_with_key):
"""Validation should use timing-safe comparison (secrets.compare_digest)."""
# This is a behavioral test: both wrong keys should take similar time
# We mainly verify the method works correctly for various inputs
assert auth_with_key.validate_master_key("test-master-key-12345") is True
assert auth_with_key.validate_master_key("test-master-key-12346") is False
def test_get_master_key(self, auth_with_key):
"""get_master_key should return the current key."""
assert auth_with_key.get_master_key() == "test-master-key-12345"
class TestProjectKeys:
"""Test project-specific key management."""
def test_add_project_key_custom(self, auth_with_key):
"""Should store a custom project key."""
auth_with_key.add_project_key("proj1", "my-project-key")
assert auth_with_key.validate_project_key("proj1", "my-project-key") is True
def test_add_project_key_generated(self, auth_with_key):
"""Should generate a key when none provided."""
key = auth_with_key.add_project_key("proj1")
assert key is not None
assert len(key) > 20
assert auth_with_key.validate_project_key("proj1", key) is True
def test_project_key_wrong_project(self, auth_with_key):
"""Project key should not work for a different project."""
auth_with_key.add_project_key("proj1", "key-for-proj1")
# proj2 has no key, so it falls back to master key
assert auth_with_key.validate_project_key("proj2", "key-for-proj1") is False
def test_fallback_to_master_key(self, auth_with_key):
"""Projects without a specific key should accept master key."""
assert auth_with_key.validate_project_key("no-key-project", "test-master-key-12345") is True
def test_remove_project_key(self, auth_with_key):
"""Removing a project key should make it fall back to master."""
auth_with_key.add_project_key("proj1", "proj-key")
auth_with_key.remove_project_key("proj1")
# After removal, should fall back to master key
assert auth_with_key.validate_project_key("proj1", "proj-key") is False
assert auth_with_key.validate_project_key("proj1", "test-master-key-12345") is True
def test_remove_nonexistent_key(self, auth_with_key):
"""Removing a key for a project that has none should not raise."""
auth_with_key.remove_project_key("nonexistent") # Should not raise
def test_has_project_key(self, auth_with_key):
"""has_project_key should reflect current state."""
assert auth_with_key.has_project_key("proj1") is False
auth_with_key.add_project_key("proj1", "key")
assert auth_with_key.has_project_key("proj1") is True
auth_with_key.remove_project_key("proj1")
assert auth_with_key.has_project_key("proj1") is False

181
tests/test_auth_logic.py Normal file
View File

@@ -0,0 +1,181 @@
#!/usr/bin/env python3
"""
Test authentication logic for different tool types
"""
def test_tool_detection():
"""Test tool type detection logic"""
SYSTEM_TOOLS = [
"list_projects",
"get_project_info",
"check_all_projects_health",
"get_project_health",
"get_system_metrics",
"get_system_uptime",
"get_rate_limit_stats",
"export_health_metrics",
"manage_api_keys_list",
"manage_api_keys_get_info",
]
test_cases = [
# (tool_name, expected_is_unified, expected_is_system, expected_skip_check, expected_project_id)
("wordpress_list_posts", True, False, True, "*"),
("woocommerce_list_products", True, False, True, "*"),
("mcp__coolify-projects__wordpress_get_post", True, False, True, "*"),
("mcp__coolify-projects__woocommerce_create_order", True, False, True, "*"),
("mcp__coolify-projects__list_projects", False, True, False, "*"),
("mcp__coolify-projects__get_system_metrics", False, True, False, "*"),
("mcp__coolify-projects__check_all_projects_health", False, True, False, "*"),
("list_projects", False, True, False, "*"),
("get_system_uptime", False, True, False, "*"),
(
"some_unknown_tool",
False,
False,
True,
"*",
), # Unknown - allow for backward compatibility
]
print("Testing tool detection logic:\n")
print(
f"{'Tool Name':<50} {'Unified':<10} {'System':<10} {'Skip Check':<12} {'Project ID':<12} {'Result'}"
)
print("-" * 110)
all_passed = True
for tool_name, exp_unified, exp_system, exp_skip, exp_project in test_cases:
# Apply the logic
is_unified_tool = (
tool_name.startswith("wordpress_")
or tool_name.startswith("woocommerce_")
or tool_name.startswith("mcp__coolify-projects__wordpress_")
or tool_name.startswith("mcp__coolify-projects__woocommerce_")
)
is_system_tool = any(tool_name.endswith(st) for st in SYSTEM_TOOLS)
if is_unified_tool:
skip_project_check = True
project_id = "*"
elif is_system_tool:
skip_project_check = False
project_id = "*"
else:
skip_project_check = True
project_id = "*"
# Check results
passed = (
is_unified_tool == exp_unified
and is_system_tool == exp_system
and skip_project_check == exp_skip
and project_id == exp_project
)
status = "✅ PASS" if passed else "❌ FAIL"
if not passed:
all_passed = False
print(
f"{tool_name:<50} {str(is_unified_tool):<10} {str(is_system_tool):<10} "
f"{str(skip_project_check):<12} {project_id:<12} {status}"
)
print("\n" + "=" * 110)
if all_passed:
print("✅ All tests passed!")
else:
print("❌ Some tests failed!")
return all_passed
def test_auth_scenarios():
"""Test authentication scenarios"""
print("\n\nTesting authentication scenarios:\n")
print(f"{'Scenario':<60} {'Expected Result':<30} {'Status'}")
print("-" * 110)
scenarios = [
# (description, key_project_id, tool_is_unified, tool_is_system, expected_result)
(
"Per-project key + Unified WP tool",
"wordpress_site1",
True,
False,
"✅ Allow (skip_project_check=True)",
),
(
"Per-project key + System tool",
"wordpress_site1",
False,
True,
"❌ Reject (key.project_id != '*')",
),
(
"Global key (*) + Unified WP tool",
"*",
True,
False,
"✅ Allow (skip_project_check=True)",
),
("Global key (*) + System tool", "*", False, True, "✅ Allow (key.project_id == '*')"),
(
"Per-project key + Unknown tool",
"wordpress_site1",
False,
False,
"✅ Allow (backward compatibility)",
),
("Global key (*) + Unknown tool", "*", False, False, "✅ Allow"),
]
for desc, key_project_id, is_unified, is_system, expected in scenarios:
# Determine skip_project_check based on tool type
if is_unified:
skip_project_check = True
elif is_system:
skip_project_check = False
else:
skip_project_check = True
# Simulate validate_key check
project_id = "*" # Always "*" in our new logic
# Check if key would be validated
if skip_project_check:
# Skip project check - key is valid regardless
would_pass_validate = True
reason = "skip_project_check=True"
else:
# Check project access
if key_project_id == "*" or key_project_id == project_id:
would_pass_validate = True
reason = f"key.project_id={key_project_id} matches project_id={project_id}"
else:
would_pass_validate = False
reason = f"key.project_id={key_project_id} != project_id={project_id}"
# Additional check for system tools
if is_system and key_project_id != "*":
would_pass_validate = False
reason = "System tools require global key (additional check)"
result = "✅ Allow" if would_pass_validate else "❌ Reject"
status = "✅ PASS" if (result in expected) else "❌ FAIL"
print(f"{desc:<60} {result:<30} {status}")
if status == "❌ FAIL":
print(f" └─ Expected: {expected}")
print(f" └─ Reason: {reason}")
print("\n" + "=" * 110)
if __name__ == "__main__":
test_tool_detection()
test_auth_scenarios()

View File

@@ -0,0 +1,270 @@
"""Tests for Community Build Sync Script."""
import tempfile
from pathlib import Path
import pytest
# Import sync module from scripts
import sys
sys.path.insert(0, str(Path(__file__).parent.parent / "scripts" / "community-build"))
from sync import (
SyncReport,
apply_branding_transform,
collect_files,
parse_communityignore,
scan_for_secrets,
should_exclude,
should_transform,
sync,
)
class TestParseCommunityignore:
"""Test .communityignore parsing."""
def test_parse_valid_file(self, tmp_path):
ignore_file = tmp_path / ".communityignore"
ignore_file.write_text("# comment\n\nfoo/\n*.pyc\nbar.txt\n")
patterns = parse_communityignore(ignore_file)
assert patterns == ["foo/", "*.pyc", "bar.txt"]
def test_parse_missing_file(self, tmp_path):
patterns = parse_communityignore(tmp_path / "nonexistent")
assert patterns == []
def test_skip_comments_and_empty(self, tmp_path):
ignore_file = tmp_path / ".communityignore"
ignore_file.write_text("# header\n\n# another\nreal_pattern\n\n")
patterns = parse_communityignore(ignore_file)
assert patterns == ["real_pattern"]
class TestShouldExclude:
"""Test file exclusion logic."""
def test_directory_pattern(self):
assert should_exclude("docs/plans/file.md", ["docs/plans/"]) is True
assert should_exclude("docs/other/file.md", ["docs/plans/"]) is False
def test_exact_match(self):
assert should_exclude("MASTER_CONTEXT.md", ["MASTER_CONTEXT.md"]) is True
assert should_exclude("README.md", ["MASTER_CONTEXT.md"]) is False
def test_glob_pattern(self):
assert should_exclude("module.pyc", ["*.pyc"]) is True
assert should_exclude("module.py", ["*.pyc"]) is False
def test_nested_directory(self):
assert should_exclude("__pycache__/module.cpython-311.pyc", ["__pycache__/"]) is True
def test_no_patterns(self):
assert should_exclude("any/file.txt", []) is False
def test_component_match(self):
"""Pattern matching against directory components."""
assert should_exclude("deep/__pycache__/file.pyc", ["__pycache__/"]) is True
class TestScanForSecrets:
"""Test secret detection."""
def test_no_secrets(self, tmp_path):
f = tmp_path / "clean.py"
f.write_text("x = 1\nprint('hello')\n")
assert scan_for_secrets(f, "clean.py") == []
def test_detects_private_key(self, tmp_path):
f = tmp_path / "bad.pem"
f.write_text("-----BEGIN PRIVATE KEY-----\ndata\n-----END PRIVATE KEY-----\n")
warnings = scan_for_secrets(f, "bad.pem")
assert len(warnings) > 0
def test_allowlisted_file_skipped(self, tmp_path):
f = tmp_path / "env.example"
f.write_text('password = "super_secret_value"\n')
assert scan_for_secrets(f, "env.example") == []
def test_detects_api_key_pattern(self, tmp_path):
f = tmp_path / "config.py"
f.write_text('api_key = "AKIAIOSFODNN7EXAMPLE12345"\n')
warnings = scan_for_secrets(f, "config.py")
assert len(warnings) > 0
class TestSync:
"""Test full sync operation."""
@pytest.fixture
def source_tree(self, tmp_path):
"""Create a mock source tree."""
src = tmp_path / "source"
src.mkdir()
# Create .communityignore
(src / ".communityignore").write_text("secret/\n*.log\nINTERNAL.md\n.communityignore\n")
# Create files that should be copied
(src / "README.md").write_text("# Public Readme")
(src / "core").mkdir()
(src / "core" / "auth.py").write_text("class AuthManager: pass")
# Create files that should be excluded
(src / "secret").mkdir()
(src / "secret" / "keys.json").write_text('{"key": "value"}')
(src / "INTERNAL.md").write_text("Internal docs")
(src / "app.log").write_text("log data")
return src
def test_dry_run(self, source_tree, tmp_path):
output = tmp_path / "output"
report = sync(source_tree, output, dry_run=True)
assert report.total_copied > 0
assert report.total_excluded > 0
assert not output.exists() # Nothing should be created
def test_actual_sync(self, source_tree, tmp_path):
output = tmp_path / "output"
report = sync(source_tree, output)
assert (output / "README.md").exists()
assert (output / "core" / "auth.py").exists()
assert not (output / "secret").exists()
assert not (output / "INTERNAL.md").exists()
assert not (output / "app.log").exists()
def test_communityignore_excluded(self, source_tree, tmp_path):
output = tmp_path / "output"
report = sync(source_tree, output)
# .communityignore itself should be excluded
assert not (output / ".communityignore").exists()
assert ".communityignore" in report.excluded
class TestSyncReport:
"""Test report generation."""
def test_summary_format(self):
report = SyncReport(
copied=["a.py", "b.py"],
excluded=["c.log"],
source_dir="/src",
output_dir="/out",
timestamp="2026-02-17 12:00:00 UTC",
)
summary = report.summary()
assert "Files copied: **2**" in summary
assert "Files excluded: **1**" in summary
def test_summary_with_warnings(self):
report = SyncReport(
copied=["a.py"],
excluded=[],
secret_warnings=["file.py: found secret"],
source_dir="/src",
output_dir="/out",
timestamp="now",
)
summary = report.summary()
assert "Secret Warnings" in summary
def test_summary_has_review_checklist(self):
report = SyncReport(
copied=["a.py"],
excluded=[],
source_dir="/src",
output_dir="/out",
timestamp="now",
)
summary = report.summary()
assert "Pre-Publish Review Checklist" in summary
assert "APPROVED FOR PUBLISH" in summary
class TestBrandingTransform:
"""Test branding transform (B.3)."""
def test_replaces_package_name(self):
content = 'name = "coolify-mcp-hub"'
transformed, changed = apply_branding_transform(content)
assert changed is True
assert 'name = "mcphub"' in transformed
def test_replaces_display_name(self):
content = "Welcome to Coolify MCP Hub"
transformed, changed = apply_branding_transform(content)
assert "MCP Hub" in transformed
assert "Coolify" not in transformed
def test_replaces_urls(self):
content = "Homepage: https://airano.ir"
transformed, changed = apply_branding_transform(content)
assert "mcphub.dev" in transformed
assert "airano.ir" not in transformed
def test_replaces_repo_urls(self):
content = "https://gitea.airano.ir/dev/coolify-mcp-hub"
transformed, changed = apply_branding_transform(content)
assert "github.com/mcphub/mcphub" in transformed
def test_replaces_email(self):
content = "Contact: gitea@airano.ir"
transformed, changed = apply_branding_transform(content)
assert "hello@mcphub.dev" in transformed
def test_strips_private_comments(self):
content = "line1\n# PRIVATE: internal note\nline3\n"
transformed, changed = apply_branding_transform(content)
assert "PRIVATE" not in transformed
assert "line1" in transformed
assert "line3" in transformed
def test_no_changes_returns_false(self):
content = "clean content with no markers"
transformed, changed = apply_branding_transform(content)
assert changed is False
def test_should_transform_pyproject(self):
assert should_transform("pyproject.toml") is True
def test_should_transform_readme(self):
assert should_transform("README.md") is True
def test_should_transform_docs(self):
assert should_transform("docs/OAUTH_GUIDE.md") is True
def test_should_transform_python(self):
assert should_transform("core/auth.py") is True
assert should_transform("server.py") is True
assert should_transform("plugins/wordpress/plugin.py") is True
def test_should_not_transform_binary(self):
assert should_transform("core/data.bin") is False
assert should_transform("images/logo.png") is False
def test_transform_applied_in_sync(self, tmp_path):
"""Full sync should apply transforms to matching files."""
src = tmp_path / "source"
src.mkdir()
(src / ".communityignore").write_text(".communityignore\n")
(src / "README.md").write_text("Welcome to Coolify MCP Hub\nhttps://airano.ir\n")
(src / "core").mkdir()
(src / "core" / "auth.py").write_text("# coolify-mcp-hub internal")
output = tmp_path / "output"
report = sync(src, output)
# README.md should be transformed
readme = (output / "README.md").read_text()
assert "MCP Hub" in readme
assert "Coolify" not in readme
assert "mcphub.dev" in readme
assert "README.md" in report.transformed
# auth.py SHOULD be transformed (Python files now in TRANSFORM_GLOBS)
auth = (output / "core" / "auth.py").read_text()
assert "mcphub" in auth # transformed

348
tests/test_dashboard.py Normal file
View File

@@ -0,0 +1,348 @@
"""Tests for Dashboard routes and auth (core/dashboard/).
Tests covering dashboard authentication, session management, rate limiting,
language detection, translations, and utility functions.
"""
import time
from datetime import UTC, datetime, timedelta
from unittest.mock import patch
import jwt
import pytest
from core.dashboard.auth import DashboardAuth, DashboardSession
from core.dashboard.routes import (
DASHBOARD_TRANSLATIONS,
PLUGIN_DISPLAY_NAMES,
detect_language,
get_plugin_display_name,
get_translations,
)
# --- Plugin Display Names ---
class TestPluginDisplayNames:
"""Test plugin name formatting."""
def test_known_plugins(self):
"""Should return proper display names for known plugins."""
assert get_plugin_display_name("wordpress") == "WordPress"
assert get_plugin_display_name("woocommerce") == "WooCommerce"
assert get_plugin_display_name("n8n") == "n8n"
assert get_plugin_display_name("gitea") == "Gitea"
assert get_plugin_display_name("supabase") == "Supabase"
assert get_plugin_display_name("openpanel") == "OpenPanel"
assert get_plugin_display_name("appwrite") == "Appwrite"
assert get_plugin_display_name("directus") == "Directus"
def test_wordpress_advanced(self):
"""Should handle underscore-separated names."""
assert get_plugin_display_name("wordpress_advanced") == "WordPress Advanced"
def test_unknown_plugin_titlecased(self):
"""Should title-case unknown plugin types."""
assert get_plugin_display_name("my_custom_plugin") == "My Custom Plugin"
def test_all_nine_plugins_mapped(self):
"""All 9 plugin types should be in the display names map."""
expected = {
"wordpress", "woocommerce", "wordpress_advanced", "gitea",
"n8n", "supabase", "openpanel", "appwrite", "directus",
}
assert expected == set(PLUGIN_DISPLAY_NAMES.keys())
# --- Language Detection ---
class TestLanguageDetection:
"""Test language detection from headers and query params."""
def test_default_english(self):
"""Should default to English."""
assert detect_language(None) == "en"
def test_detect_farsi_from_header(self):
"""Should detect Farsi from Accept-Language header."""
assert detect_language("fa-IR,fa;q=0.9,en;q=0.8") == "fa"
def test_detect_farsi_short(self):
"""Should detect 'fa' in Accept-Language."""
assert detect_language("fa") == "fa"
def test_query_param_overrides_header(self):
"""Query parameter should override Accept-Language header."""
assert detect_language("en-US", query_lang="fa") == "fa"
def test_query_param_english(self):
"""Should respect English query parameter."""
assert detect_language("fa-IR", query_lang="en") == "en"
def test_invalid_query_lang_ignored(self):
"""Invalid query lang should be ignored, fall back to header."""
assert detect_language("fa-IR", query_lang="de") == "fa"
def test_english_header(self):
"""Should return English for English header."""
assert detect_language("en-US,en;q=0.9") == "en"
# --- Translations ---
class TestTranslations:
"""Test translation system."""
def test_english_translations_exist(self):
"""English translations should be available."""
trans = get_translations("en")
assert trans["dashboard"] == "Dashboard"
assert trans["projects"] == "Projects"
assert trans["login_title"] == "Dashboard Login"
def test_farsi_translations_exist(self):
"""Farsi translations should be available."""
trans = get_translations("fa")
assert trans["dashboard"] == "داشبورد"
assert trans["projects"] == "پروژه‌ها"
assert trans["login_title"] == "ورود به داشبورد"
def test_unknown_lang_falls_back_to_english(self):
"""Unknown language should fall back to English."""
trans = get_translations("de")
assert trans["dashboard"] == "Dashboard"
def test_both_languages_have_same_keys(self):
"""English and Farsi should have the same translation keys."""
en_keys = set(DASHBOARD_TRANSLATIONS["en"].keys())
fa_keys = set(DASHBOARD_TRANSLATIONS["fa"].keys())
assert en_keys == fa_keys, f"Missing keys in fa: {en_keys - fa_keys}, extra in fa: {fa_keys - en_keys}"
def test_no_empty_translations(self):
"""No translation value should be empty."""
for lang, translations in DASHBOARD_TRANSLATIONS.items():
for key, value in translations.items():
assert value.strip() != "", f"Empty translation: {lang}.{key}"
# --- Dashboard Auth ---
class TestDashboardAuthInit:
"""Test DashboardAuth initialization."""
def test_init_with_explicit_key(self):
"""Should use explicit secret key."""
auth = DashboardAuth(secret_key="test-secret", master_api_key="sk-master")
assert auth.secret_key == "test-secret"
assert auth.master_api_key == "sk-master"
def test_init_generates_random_key(self, monkeypatch):
"""Should generate random key when none provided."""
monkeypatch.delenv("DASHBOARD_SESSION_SECRET", raising=False)
monkeypatch.delenv("OAUTH_JWT_SECRET_KEY", raising=False)
auth = DashboardAuth(master_api_key="sk-master")
assert auth.secret_key is not None
assert len(auth.secret_key) == 64 # hex of 32 bytes
def test_default_session_expiry(self):
"""Default session expiry should be 24 hours."""
auth = DashboardAuth(secret_key="test", master_api_key="sk-master")
assert auth.session_expiry_hours == 24
def test_custom_session_expiry(self, monkeypatch):
"""Should respect DASHBOARD_SESSION_EXPIRY_HOURS env var."""
monkeypatch.setenv("DASHBOARD_SESSION_EXPIRY_HOURS", "48")
auth = DashboardAuth(secret_key="test", master_api_key="sk-master")
assert auth.session_expiry_hours == 48
class TestDashboardAuthValidation:
"""Test API key validation for dashboard login."""
@pytest.fixture
def auth(self):
return DashboardAuth(secret_key="test-secret-key", master_api_key="sk-master-key-123")
def test_valid_master_key(self, auth):
"""Should accept valid master API key."""
is_valid, user_type, key_id = auth.validate_api_key("sk-master-key-123")
assert is_valid is True
assert user_type == "master"
assert key_id is None
def test_invalid_key_rejected(self, auth):
"""Should reject invalid API key."""
is_valid, user_type, key_id = auth.validate_api_key("wrong-key")
assert is_valid is False
assert user_type == ""
def test_empty_key_rejected(self, auth):
"""Should reject empty API key."""
is_valid, user_type, key_id = auth.validate_api_key("")
assert is_valid is False
def test_none_key_rejected(self, auth):
"""Should reject None API key."""
is_valid, user_type, key_id = auth.validate_api_key(None)
assert is_valid is False
class TestDashboardRateLimiting:
"""Test login rate limiting."""
@pytest.fixture
def auth(self):
return DashboardAuth(
secret_key="test-secret", master_api_key="sk-master",
)
def test_within_limit(self, auth):
"""Should allow requests within rate limit."""
assert auth.check_rate_limit("192.168.1.1") is True
def test_exceeds_limit(self, auth):
"""Should block after exceeding rate limit."""
ip = "192.168.1.100"
# Record max_login_attempts (default 5)
for _ in range(auth.max_login_attempts):
auth.record_login_attempt(ip)
assert auth.check_rate_limit(ip) is False
def test_different_ips_independent(self, auth):
"""Different IPs should have independent rate limits."""
ip1 = "10.0.0.1"
ip2 = "10.0.0.2"
for _ in range(auth.max_login_attempts):
auth.record_login_attempt(ip1)
assert auth.check_rate_limit(ip1) is False
assert auth.check_rate_limit(ip2) is True
def test_attempts_expire(self, auth):
"""Old attempts should expire after 1 minute window."""
ip = "10.0.0.50"
# Manually add old timestamps
old_time = datetime.now(UTC) - timedelta(minutes=2)
auth._login_attempts[ip] = [old_time] * 10
# Old attempts should be cleaned up
assert auth.check_rate_limit(ip) is True
class TestDashboardSessionManagement:
"""Test session creation and validation."""
@pytest.fixture
def auth(self):
return DashboardAuth(secret_key="test-session-secret", master_api_key="sk-master")
def test_create_session_returns_token(self, auth):
"""Should create a JWT session token."""
token = auth.create_session("master")
assert isinstance(token, str)
assert len(token) > 0
def test_validate_valid_session(self, auth):
"""Should validate a freshly created session."""
token = auth.create_session("master")
session = auth.validate_session(token)
assert session is not None
assert isinstance(session, DashboardSession)
assert session.user_type == "master"
assert session.session_id is not None
def test_session_contains_key_id(self, auth):
"""API key sessions should include key_id."""
token = auth.create_session("api_key", key_id="key_abc123")
session = auth.validate_session(token)
assert session.key_id == "key_abc123"
def test_master_session_no_key_id(self, auth):
"""Master sessions should have no key_id."""
token = auth.create_session("master")
session = auth.validate_session(token)
assert session.key_id is None
def test_expired_session_rejected(self, auth):
"""Should reject expired sessions."""
# Create a token that's already expired
now = datetime.now(UTC)
payload = {
"sid": "test-session",
"type": "master",
"iat": (now - timedelta(hours=48)).timestamp(),
"exp": (now - timedelta(hours=1)).timestamp(),
}
token = jwt.encode(payload, auth.secret_key, algorithm="HS256")
session = auth.validate_session(token)
assert session is None
def test_invalid_token_rejected(self, auth):
"""Should reject malformed tokens."""
assert auth.validate_session("not-a-jwt-token") is None
def test_wrong_secret_rejected(self, auth):
"""Should reject tokens signed with different secret."""
token = jwt.encode(
{"sid": "x", "type": "master", "iat": time.time(), "exp": time.time() + 3600},
"wrong-secret",
algorithm="HS256",
)
assert auth.validate_session(token) is None
def test_empty_token_rejected(self, auth):
"""Should handle empty token."""
assert auth.validate_session("") is None
assert auth.validate_session(None) is None
def test_session_expiry_matches_config(self, auth):
"""Session expiry should match configured hours."""
token = auth.create_session("master")
session = auth.validate_session(token)
# validate_session uses datetime.fromtimestamp() which returns local time (naive)
# So we compare with local datetime.now() (also naive)
expected_expiry = datetime.now() + timedelta(hours=24)
assert abs((session.expires_at - expected_expiry).total_seconds()) < 5
class TestDashboardCookieManagement:
"""Test session cookie handling."""
@pytest.fixture
def auth(self):
return DashboardAuth(secret_key="cookie-secret", master_api_key="sk-master")
def test_set_session_cookie(self, auth):
"""Should set httpOnly cookie on response."""
from starlette.responses import Response
response = Response("OK")
token = auth.create_session("master")
auth.set_session_cookie(response, token)
# Verify cookie was set (check raw headers)
cookie_header = None
for key, value in response.raw_headers:
if key == b"set-cookie":
cookie_header = value.decode()
break
assert cookie_header is not None
assert "mcp_dashboard_session=" in cookie_header
assert "httponly" in cookie_header.lower()
def test_clear_session_cookie(self, auth):
"""Should clear session cookie."""
from starlette.responses import Response
response = Response("OK")
auth.clear_session_cookie(response)
cookie_header = None
for key, value in response.raw_headers:
if key == b"set-cookie":
cookie_header = value.decode()
break
assert cookie_header is not None
assert "mcp_dashboard_session=" in cookie_header

View File

@@ -0,0 +1,74 @@
import tempfile
import pytest
from core.oauth.client_registry import ClientRegistry
@pytest.fixture
def registry():
"""Create temporary client registry for tests"""
with tempfile.TemporaryDirectory() as tmpdir:
yield ClientRegistry(tmpdir)
def test_create_client(registry):
"""Test client creation"""
client_id, client_secret = registry.create_client(
client_name="Test App", redirect_uris=["http://localhost:3000/callback"]
)
# Check client_id format
assert client_id.startswith("cmp_client_")
# Check secret is generated
assert len(client_secret) > 20
# Retrieve client
client = registry.get_client(client_id)
assert client is not None
assert client.client_name == "Test App"
assert "authorization_code" in client.grant_types
def test_validate_client_secret(registry):
"""Test client secret validation"""
client_id, client_secret = registry.create_client(
client_name="Test App", redirect_uris=["http://localhost:3000/callback"]
)
# Valid secret
assert registry.validate_client_secret(client_id, client_secret)
# Invalid secret
assert not registry.validate_client_secret(client_id, "wrong_secret")
# Non-existent client
assert not registry.validate_client_secret("fake_id", client_secret)
def test_list_clients(registry):
"""Test listing clients"""
# Create multiple clients
registry.create_client("App 1", ["http://localhost:3001/callback"])
registry.create_client("App 2", ["http://localhost:3002/callback"])
clients = registry.list_clients()
assert len(clients) == 2
assert clients[0].client_name in ["App 1", "App 2"]
def test_delete_client(registry):
"""Test client deletion"""
client_id, _ = registry.create_client(
client_name="Test App", redirect_uris=["http://localhost:3000/callback"]
)
# Delete
assert registry.delete_client(client_id)
# Should not exist
assert registry.get_client(client_id) is None
# Delete non-existent
assert not registry.delete_client("fake_id")

View File

@@ -0,0 +1,339 @@
"""
OAuth 2.1 Integration Tests
Tests the complete OAuth flow end-to-end:
1. Client registration
2. Authorization request
3. Authorization code generation
4. Token exchange
5. Token refresh
6. JWT validation
"""
import os
import tempfile
import pytest
from core.oauth import (
OAuthError,
generate_code_challenge,
generate_code_verifier,
get_client_registry,
get_oauth_server,
get_storage,
get_token_manager,
)
@pytest.fixture
def temp_storage():
"""Create temporary storage for tests"""
with tempfile.TemporaryDirectory() as tmpdir:
# Set environment for storage
os.environ["OAUTH_STORAGE_PATH"] = tmpdir
os.environ["OAUTH_JWT_SECRET_KEY"] = "test_secret_key_for_integration_tests"
# Reset singletons by clearing them
from core.oauth import client_registry, server, storage, token_manager
client_registry._client_registry = None
token_manager._token_manager = None
storage._storage = None
server._oauth_server = None
yield tmpdir
@pytest.fixture
def oauth_components(temp_storage):
"""Get fresh OAuth components"""
return {
"server": get_oauth_server(),
"client_registry": get_client_registry(),
"token_manager": get_token_manager(),
"storage": get_storage(),
}
@pytest.fixture
def test_client(oauth_components):
"""Create a test OAuth client"""
client_registry = oauth_components["client_registry"]
client_id, client_secret = client_registry.create_client(
client_name="Test Client",
redirect_uris=["http://localhost:3000/callback"],
grant_types=["authorization_code", "refresh_token", "client_credentials"],
allowed_scopes=["read", "write", "admin"],
)
return {
"client_id": client_id,
"client_secret": client_secret,
"redirect_uri": "http://localhost:3000/callback",
}
def test_full_authorization_code_flow(oauth_components, test_client):
"""
Test complete Authorization Code flow with PKCE
Steps:
1. Generate PKCE verifier/challenge
2. Validate authorization request
3. Create authorization code
4. Exchange code for tokens
5. Validate access token
6. Refresh access token
"""
server = oauth_components["server"]
token_manager = oauth_components["token_manager"]
# Step 1: Generate PKCE
code_verifier = generate_code_verifier()
code_challenge = generate_code_challenge(code_verifier)
# Step 2: Validate authorization request
validated = server.validate_authorization_request(
client_id=test_client["client_id"],
redirect_uri=test_client["redirect_uri"],
response_type="code",
code_challenge=code_challenge,
code_challenge_method="S256",
scope="read write",
state="random_state_token",
)
assert validated["client_id"] == test_client["client_id"]
assert validated["scope"] == "read write"
assert validated["state"] == "random_state_token"
# Step 3: Create authorization code
auth_code = server.create_authorization_code(
client_id=validated["client_id"],
redirect_uri=validated["redirect_uri"],
scope=validated["scope"],
code_challenge=validated["code_challenge"],
code_challenge_method=validated["code_challenge_method"],
)
assert auth_code.startswith("auth_")
# Step 4: Exchange code for tokens
token_response = server.exchange_code_for_tokens(
client_id=test_client["client_id"],
client_secret=test_client["client_secret"],
code=auth_code,
redirect_uri=test_client["redirect_uri"],
code_verifier=code_verifier,
)
assert token_response.access_token
assert token_response.refresh_token
assert token_response.token_type == "Bearer"
assert token_response.expires_in > 0
assert token_response.scope == "read write"
# Step 5: Validate access token
payload = token_manager.validate_access_token(token_response.access_token)
assert payload["client_id"] == test_client["client_id"]
assert payload["scope"] == "read write"
assert "exp" in payload
assert "iat" in payload
assert "jti" in payload
# Step 6: Refresh access token
new_tokens = server.handle_refresh_token_grant(
client_id=test_client["client_id"],
client_secret=test_client["client_secret"],
refresh_token=token_response.refresh_token,
)
assert new_tokens.access_token != token_response.access_token
assert new_tokens.refresh_token != token_response.refresh_token
assert new_tokens.scope == "read write"
def test_client_credentials_flow(oauth_components, test_client):
"""
Test Client Credentials flow (machine-to-machine)
"""
server = oauth_components["server"]
token_manager = oauth_components["token_manager"]
# Request token with client credentials
token_response = server.handle_client_credentials_grant(
client_id=test_client["client_id"], client_secret=test_client["client_secret"], scope="read"
)
assert token_response.access_token
assert token_response.refresh_token is None # No refresh token for client credentials
assert token_response.scope == "read"
# Validate token
payload = token_manager.validate_access_token(token_response.access_token)
assert payload["client_id"] == test_client["client_id"]
assert payload["scope"] == "read"
def test_pkce_validation_failure(oauth_components, test_client):
"""Test that PKCE validation fails with wrong code_verifier"""
server = oauth_components["server"]
# Generate PKCE
code_verifier = generate_code_verifier()
code_challenge = generate_code_challenge(code_verifier)
# Create authorization code
validated = server.validate_authorization_request(
client_id=test_client["client_id"],
redirect_uri=test_client["redirect_uri"],
response_type="code",
code_challenge=code_challenge,
code_challenge_method="S256",
)
auth_code = server.create_authorization_code(
client_id=validated["client_id"],
redirect_uri=validated["redirect_uri"],
scope=validated["scope"],
code_challenge=validated["code_challenge"],
code_challenge_method=validated["code_challenge_method"],
)
# Try to exchange with WRONG code_verifier
wrong_verifier = generate_code_verifier()
with pytest.raises(OAuthError, match="PKCE validation failed"):
server.exchange_code_for_tokens(
client_id=test_client["client_id"],
client_secret=test_client["client_secret"],
code=auth_code,
redirect_uri=test_client["redirect_uri"],
code_verifier=wrong_verifier, # WRONG!
)
def test_authorization_code_reuse_detection(oauth_components, test_client):
"""Test that reusing an authorization code is detected"""
server = oauth_components["server"]
# Generate PKCE and authorization code
code_verifier = generate_code_verifier()
code_challenge = generate_code_challenge(code_verifier)
validated = server.validate_authorization_request(
client_id=test_client["client_id"],
redirect_uri=test_client["redirect_uri"],
response_type="code",
code_challenge=code_challenge,
code_challenge_method="S256",
)
auth_code = server.create_authorization_code(
client_id=validated["client_id"],
redirect_uri=validated["redirect_uri"],
scope=validated["scope"],
code_challenge=validated["code_challenge"],
code_challenge_method=validated["code_challenge_method"],
)
# First exchange - should succeed
token_response = server.exchange_code_for_tokens(
client_id=test_client["client_id"],
client_secret=test_client["client_secret"],
code=auth_code,
redirect_uri=test_client["redirect_uri"],
code_verifier=code_verifier,
)
assert token_response.access_token
# Second exchange with same code - should fail
with pytest.raises(OAuthError, match="already used"):
server.exchange_code_for_tokens(
client_id=test_client["client_id"],
client_secret=test_client["client_secret"],
code=auth_code,
redirect_uri=test_client["redirect_uri"],
code_verifier=code_verifier,
)
def test_invalid_client_credentials(oauth_components, test_client):
"""Test that invalid client credentials are rejected"""
server = oauth_components["server"]
# Try with wrong client secret
with pytest.raises(OAuthError, match="Invalid client credentials"):
server.handle_client_credentials_grant(
client_id=test_client["client_id"], client_secret="wrong_secret", scope="read"
)
def test_scope_validation(oauth_components, test_client):
"""Test that invalid scopes are rejected"""
server = oauth_components["server"]
# Try to request invalid scope
code_verifier = generate_code_verifier()
code_challenge = generate_code_challenge(code_verifier)
with pytest.raises(OAuthError, match="not allowed"):
server.validate_authorization_request(
client_id=test_client["client_id"],
redirect_uri=test_client["redirect_uri"],
response_type="code",
code_challenge=code_challenge,
code_challenge_method="S256",
scope="invalid_scope", # Not in allowed_scopes
)
def test_redirect_uri_validation(oauth_components, test_client):
"""Test that invalid redirect URIs are rejected"""
server = oauth_components["server"]
code_verifier = generate_code_verifier()
code_challenge = generate_code_challenge(code_verifier)
# Try with unregistered redirect_uri
with pytest.raises(OAuthError, match="Invalid redirect_uri"):
server.validate_authorization_request(
client_id=test_client["client_id"],
redirect_uri="http://evil.com/callback", # Not registered!
response_type="code",
code_challenge=code_challenge,
code_challenge_method="S256",
)
def test_expired_token(oauth_components, test_client):
"""Test that expired JWT tokens are rejected"""
import time
import jwt as pyjwt
token_manager = oauth_components["token_manager"]
# Set very short TTL
original_ttl = token_manager.access_token_ttl
token_manager.access_token_ttl = 1 # 1 second
# Generate token
access_token = token_manager.generate_access_token(
client_id=test_client["client_id"], scope="read"
)
# Wait for expiration
time.sleep(2)
# Try to validate expired token
with pytest.raises(pyjwt.ExpiredSignatureError):
token_manager.validate_access_token(access_token)
# Restore original TTL
token_manager.access_token_ttl = original_ttl

50
tests/test_oauth_pkce.py Normal file
View File

@@ -0,0 +1,50 @@
import pytest
from core.oauth.pkce import generate_code_challenge, generate_code_verifier, validate_code_challenge
def test_code_verifier_generation():
"""Test code verifier generation"""
verifier = generate_code_verifier()
# Length check
assert 43 <= len(verifier) <= 128
# Character set check
import re
assert re.match(r"^[A-Za-z0-9_-]+$", verifier)
# Uniqueness
assert verifier != generate_code_verifier()
def test_code_challenge_s256():
"""Test S256 code challenge generation"""
# RFC 7636 Appendix B example
verifier = "dBjftJeZ4CVP-mB92K27uhbUJU1p1r_wW1gFWFOEjXk"
expected = "E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM"
challenge = generate_code_challenge(verifier, "S256")
assert challenge == expected
def test_code_challenge_validation():
"""Test PKCE validation"""
verifier = generate_code_verifier()
challenge = generate_code_challenge(verifier)
# Valid
assert validate_code_challenge(verifier, challenge, "S256")
# Invalid verifier
assert not validate_code_challenge("wrong", challenge, "S256")
# Invalid challenge
assert not validate_code_challenge(verifier, "wrong", "S256")
def test_plain_method_not_allowed():
"""Test that plain method is not allowed (OAuth 2.1)"""
with pytest.raises(ValueError, match="S256"):
generate_code_challenge("verifier", "plain")

View File

@@ -0,0 +1,44 @@
from datetime import UTC, datetime, timedelta
import pytest
from core.oauth.schemas import AuthorizationCode, OAuthClient
def test_oauth_client_validation():
"""Test OAuth client validation"""
client = OAuthClient(
client_id="cmp_client_test",
client_secret_hash="sha256_hash",
client_name="Test Client",
redirect_uris=["http://localhost:3000/callback"],
)
assert client.client_id == "cmp_client_test"
assert "authorization_code" in client.grant_types
assert "read" in client.allowed_scopes
def test_invalid_redirect_uri():
"""Test invalid redirect URI"""
with pytest.raises(ValueError, match="Invalid redirect URI"):
OAuthClient(
client_id="test",
client_secret_hash="hash",
client_name="Test",
redirect_uris=["invalid-uri"], # Missing http://
)
def test_authorization_code_expiry():
"""Test authorization code expiry"""
code = AuthorizationCode(
code="test_code",
client_id="client1",
redirect_uri="http://localhost:3000/callback",
scope="read",
code_challenge="challenge",
expires_at=datetime.now(UTC) - timedelta(seconds=1), # Expired
)
assert code.is_expired()

View File

@@ -0,0 +1,71 @@
import tempfile
from datetime import UTC, datetime, timedelta
import pytest
from core.oauth.schemas import AuthorizationCode, RefreshToken
from core.oauth.storage import JSONStorage
@pytest.fixture
def storage():
"""Create temporary storage for tests"""
with tempfile.TemporaryDirectory() as tmpdir:
yield JSONStorage(tmpdir)
def test_save_and_get_authorization_code(storage):
"""Test authorization code storage"""
code = AuthorizationCode(
code="test_code_123",
client_id="client_1",
redirect_uri="http://localhost:3000/callback",
scope="read write",
code_challenge="challenge_123",
expires_at=datetime.now(UTC) + timedelta(minutes=5),
)
# Save
assert storage.save_authorization_code(code)
# Get
retrieved = storage.get_authorization_code("test_code_123")
assert retrieved is not None
assert retrieved.code == "test_code_123"
assert retrieved.client_id == "client_1"
def test_expired_code_auto_cleanup(storage):
"""Test expired code auto-cleanup"""
code = AuthorizationCode(
code="expired_code",
client_id="client_1",
redirect_uri="http://localhost:3000/callback",
scope="read",
code_challenge="challenge",
expires_at=datetime.now(UTC) - timedelta(seconds=1), # Expired
)
storage.save_authorization_code(code)
# Should return None and cleanup
retrieved = storage.get_authorization_code("expired_code")
assert retrieved is None
def test_refresh_token_revocation(storage):
"""Test refresh token revocation"""
token = RefreshToken(
token="refresh_token_123",
client_id="client_1",
expires_at=datetime.now(UTC) + timedelta(days=7),
)
storage.save_refresh_token(token)
# Revoke
assert storage.revoke_refresh_token("refresh_token_123")
# Should return None
retrieved = storage.get_refresh_token("refresh_token_123")
assert retrieved is None

View File

@@ -0,0 +1,78 @@
import os
import tempfile
import jwt
import pytest
from core.oauth.token_manager import SecurityError, TokenManager
@pytest.fixture
def token_manager():
os.environ["OAUTH_JWT_SECRET_KEY"] = "test_secret_key"
# Use temporary directory for storage
with tempfile.TemporaryDirectory() as tmpdir:
os.environ["OAUTH_STORAGE_PATH"] = tmpdir
yield TokenManager()
def test_generate_access_token(token_manager):
"""Test access token generation"""
token = token_manager.generate_access_token(
client_id="test_client", scope="read write", user_id="user_123"
)
# Should be valid JWT
assert len(token.split(".")) == 3
# Validate
payload = token_manager.validate_access_token(token)
assert payload["client_id"] == "test_client"
assert payload["scope"] == "read write"
assert payload["sub"] == "user_123"
def test_access_token_expiry(token_manager):
"""Test expired token"""
# Generate with short TTL
token_manager.access_token_ttl = 1
token = token_manager.generate_access_token("client", "read")
import time
time.sleep(2)
# Should raise
with pytest.raises(jwt.ExpiredSignatureError):
token_manager.validate_access_token(token)
def test_refresh_token_rotation(token_manager):
"""Test refresh token rotation"""
# Generate initial tokens
access_token = token_manager.generate_access_token("client", "read")
refresh_token = token_manager.generate_refresh_token("client", access_token)
# Rotate
new_tokens = token_manager.rotate_refresh_token(refresh_token, "client")
assert "access_token" in new_tokens
assert "refresh_token" in new_tokens
assert new_tokens["refresh_token"] != refresh_token
# Old token should be revoked - reuse triggers SecurityError
with pytest.raises(SecurityError, match="reuse detected"):
token_manager.rotate_refresh_token(refresh_token, "client")
def test_refresh_token_reuse_detection(token_manager):
"""Test reuse detection"""
access_token = token_manager.generate_access_token("client", "read")
refresh_token = token_manager.generate_refresh_token("client", access_token)
# First rotation
token_manager.rotate_refresh_token(refresh_token, "client")
# Attempt reuse
with pytest.raises(SecurityError, match="reuse detected"):
token_manager.rotate_refresh_token(refresh_token, "client")

166
tests/test_rate_limiter.py Normal file
View File

@@ -0,0 +1,166 @@
"""Tests for Rate Limiter (core/rate_limiter.py)."""
import time
import pytest
from core.rate_limiter import ClientRateLimitState, RateLimitConfig, RateLimiter, TokenBucket
class TestRateLimitConfig:
"""Test rate limit configuration."""
def test_defaults(self):
config = RateLimitConfig()
assert config.per_minute == 60
assert config.per_hour == 1000
assert config.per_day == 10000
def test_custom_values(self):
config = RateLimitConfig(per_minute=10, per_hour=100, per_day=500)
assert config.per_minute == 10
def test_from_env(self, monkeypatch):
monkeypatch.setenv("RATE_LIMIT_PER_MINUTE", "30")
monkeypatch.setenv("RATE_LIMIT_PER_HOUR", "500")
monkeypatch.setenv("RATE_LIMIT_PER_DAY", "5000")
config = RateLimitConfig.from_env()
assert config.per_minute == 30
assert config.per_hour == 500
assert config.per_day == 5000
def test_from_env_with_prefix(self, monkeypatch):
monkeypatch.setenv("WP_RATE_LIMIT_PER_MINUTE", "20")
config = RateLimitConfig.from_env("WP")
assert config.per_minute == 20
class TestTokenBucket:
"""Test token bucket algorithm."""
def test_initial_capacity(self):
bucket = TokenBucket(capacity=10, refill_rate=1.0)
assert bucket.get_available_tokens() == 10
def test_consume_success(self):
bucket = TokenBucket(capacity=10, refill_rate=1.0)
assert bucket.consume(1) is True
assert bucket.get_available_tokens() == 9
def test_consume_multiple(self):
bucket = TokenBucket(capacity=10, refill_rate=1.0)
assert bucket.consume(5) is True
assert bucket.get_available_tokens() == 5
def test_consume_exceeds_capacity(self):
bucket = TokenBucket(capacity=5, refill_rate=1.0)
assert bucket.consume(6) is False
# Tokens should not be consumed on failure
assert bucket.get_available_tokens() == 5
def test_consume_until_empty(self):
bucket = TokenBucket(capacity=3, refill_rate=0.0) # no refill
assert bucket.consume(1) is True
assert bucket.consume(1) is True
assert bucket.consume(1) is True
assert bucket.consume(1) is False
def test_get_wait_time_when_available(self):
bucket = TokenBucket(capacity=10, refill_rate=1.0)
assert bucket.get_wait_time(1) == 0.0
def test_get_wait_time_when_empty(self):
bucket = TokenBucket(capacity=1, refill_rate=1.0)
bucket.consume(1)
wait = bucket.get_wait_time(1)
assert wait > 0
class TestClientRateLimitState:
"""Test per-client rate limit state."""
def _make_client(self, per_minute=5, per_hour=100, per_day=1000):
return ClientRateLimitState(
client_id="test-client",
minute_bucket=TokenBucket(capacity=per_minute, refill_rate=per_minute / 60.0),
hour_bucket=TokenBucket(capacity=per_hour, refill_rate=per_hour / 3600.0),
day_bucket=TokenBucket(capacity=per_day, refill_rate=per_day / 86400.0),
)
def test_allowed_request(self):
client = self._make_client()
allowed, reason, retry_after = client.check_and_consume()
assert allowed is True
assert reason == ""
assert retry_after == 0.0
def test_minute_limit_exceeded(self):
client = self._make_client(per_minute=2)
client.check_and_consume()
client.check_and_consume()
allowed, reason, retry_after = client.check_and_consume()
assert allowed is False
assert "per minute" in reason
def test_stats(self):
client = self._make_client()
client.check_and_consume()
client.check_and_consume()
stats = client.get_stats()
assert stats["client_id"] == "test-client"
assert stats["total_requests"] == 2
class TestRateLimiter:
"""Test the main RateLimiter class."""
@pytest.fixture
def limiter(self):
return RateLimiter()
def test_first_request_allowed(self, limiter):
allowed, msg, retry = limiter.check_rate_limit("client1")
assert allowed is True
def test_different_clients_independent(self, limiter):
for _ in range(50):
limiter.check_rate_limit("client1")
allowed, _, _ = limiter.check_rate_limit("client2")
assert allowed is True
def test_get_client_stats(self, limiter):
limiter.check_rate_limit("client1")
stats = limiter.get_client_stats("client1")
assert stats is not None
assert stats["client_id"] == "client1"
def test_get_client_stats_unknown(self, limiter):
assert limiter.get_client_stats("unknown") is None
def test_reset_client(self, limiter):
limiter.check_rate_limit("client1")
assert limiter.reset_client("client1") is True
assert limiter.get_client_stats("client1") is None
def test_reset_nonexistent_client(self, limiter):
assert limiter.reset_client("nonexistent") is False
def test_reset_all(self, limiter):
limiter.check_rate_limit("client1")
limiter.check_rate_limit("client2")
count = limiter.reset_all()
assert count == 2
assert limiter.get_client_stats("client1") is None
def test_get_all_stats(self, limiter):
limiter.check_rate_limit("client1")
stats = limiter.get_all_stats()
assert "global" in stats
assert "default_limits" in stats
assert stats["global"]["total_requests"] >= 1
assert stats["global"]["active_clients"] == 1
def test_configure_limits(self, limiter):
limiter.configure_limits("n8n", per_minute=10, per_hour=50)
assert limiter.plugin_configs["n8n"].per_minute == 10
assert limiter.plugin_configs["n8n"].per_hour == 50

239
tests/test_site_manager.py Normal file
View File

@@ -0,0 +1,239 @@
"""Tests for Site Manager (core/site_manager.py)."""
import pytest
from core.site_manager import SiteConfig, SiteManager
# --- SiteConfig Tests ---
class TestSiteConfig:
"""Test SiteConfig Pydantic model."""
def test_basic_creation(self):
"""Should create a config with required fields."""
config = SiteConfig(
site_id="site1",
plugin_type="wordpress",
url="https://example.com",
)
assert config.site_id == "site1"
assert config.plugin_type == "wordpress"
assert config.url == "https://example.com"
def test_alias_none_when_not_provided(self):
"""Alias should be None when not explicitly provided."""
config = SiteConfig(site_id="site1", plugin_type="wordpress")
# Pydantic V2: field_validator doesn't run on default values
# get_display_name() handles the fallback to site_id
assert config.alias is None
assert config.get_display_name() == "site1"
def test_custom_alias(self):
"""Custom alias should override default."""
config = SiteConfig(site_id="site1", plugin_type="wordpress", alias="myblog")
assert config.alias == "myblog"
def test_get_full_id(self):
"""Full ID should be plugin_type_site_id."""
config = SiteConfig(site_id="site1", plugin_type="wordpress")
assert config.get_full_id() == "wordpress_site1"
def test_get_display_name_with_alias(self):
"""Display name should prefer alias."""
config = SiteConfig(site_id="site1", plugin_type="wordpress", alias="myblog")
assert config.get_display_name() == "myblog"
def test_get_display_name_without_alias(self):
"""Display name should fall back to site_id."""
config = SiteConfig(site_id="site1", plugin_type="wordpress", alias=None)
assert config.get_display_name() == "site1"
def test_extra_fields_allowed(self):
"""Plugin-specific fields should be accepted."""
config = SiteConfig(
site_id="site1",
plugin_type="wordpress",
consumer_key="ck_123",
consumer_secret="cs_456",
)
assert config.model_extra["consumer_key"] == "ck_123"
def test_to_dict(self):
"""to_dict should include all fields."""
config = SiteConfig(
site_id="site1",
plugin_type="wordpress",
url="https://example.com",
)
d = config.to_dict()
assert d["site_id"] == "site1"
assert d["plugin_type"] == "wordpress"
assert d["url"] == "https://example.com"
# --- SiteManager Tests ---
class TestSiteManagerRegistration:
"""Test site registration and lookup."""
@pytest.fixture
def manager(self):
return SiteManager()
def test_register_site(self, manager):
"""Should register a site and retrieve it by ID."""
config = SiteConfig(site_id="site1", plugin_type="wordpress", url="https://example.com")
manager.register_site(config)
result = manager.get_site_config("wordpress", "site1")
assert result.url == "https://example.com"
def test_register_site_with_alias(self, manager):
"""Should be retrievable by alias."""
config = SiteConfig(
site_id="site1", plugin_type="wordpress", alias="myblog", url="https://example.com"
)
manager.register_site(config)
result = manager.get_site_config("wordpress", "myblog")
assert result.site_id == "site1"
def test_site_not_found_raises(self, manager):
"""Should raise ValueError for unknown site."""
config = SiteConfig(site_id="site1", plugin_type="wordpress")
manager.register_site(config)
with pytest.raises(ValueError, match="not configured"):
manager.get_site_config("wordpress", "nonexistent")
def test_unknown_plugin_type_raises(self, manager):
"""Should raise ValueError for unknown plugin type."""
with pytest.raises(ValueError, match="No sites configured"):
manager.get_site_config("unknown_plugin", "site1")
def test_list_sites(self, manager):
"""Should list all site IDs and aliases."""
config1 = SiteConfig(site_id="site1", plugin_type="wordpress", alias="blog1")
config2 = SiteConfig(site_id="site2", plugin_type="wordpress", alias="blog2")
manager.register_site(config1)
manager.register_site(config2)
sites = manager.list_sites("wordpress")
assert "site1" in sites
assert "site2" in sites
assert "blog1" in sites
assert "blog2" in sites
def test_list_sites_empty_plugin(self, manager):
"""Should return empty list for unknown plugin type."""
assert manager.list_sites("nonexistent") == []
class TestSiteManagerDiscovery:
"""Test site discovery from environment variables."""
def test_discover_wordpress_sites(self, monkeypatch):
"""Should discover sites from WORDPRESS_SITE1_* env vars."""
monkeypatch.setenv("WORDPRESS_SITE1_URL", "https://wp1.example.com")
monkeypatch.setenv("WORDPRESS_SITE1_USERNAME", "admin")
monkeypatch.setenv("WORDPRESS_SITE1_APP_PASSWORD", "xxxx")
monkeypatch.setenv("WORDPRESS_SITE1_ALIAS", "myblog")
manager = SiteManager()
count = manager.discover_sites(["wordpress"])
assert count == 1
config = manager.get_site_config("wordpress", "site1")
assert config.url == "https://wp1.example.com"
assert config.username == "admin"
assert config.alias == "myblog"
def test_discover_multiple_sites(self, monkeypatch):
"""Should discover multiple sites for same plugin type."""
monkeypatch.setenv("WORDPRESS_SITE1_URL", "https://wp1.example.com")
monkeypatch.setenv("WORDPRESS_SITE1_USERNAME", "admin")
monkeypatch.setenv("WORDPRESS_SITE2_URL", "https://wp2.example.com")
monkeypatch.setenv("WORDPRESS_SITE2_USERNAME", "editor")
manager = SiteManager()
count = manager.discover_sites(["wordpress"])
assert count == 2
def test_discover_across_plugin_types(self, monkeypatch):
"""Should discover sites across different plugin types."""
monkeypatch.setenv("WORDPRESS_SITE1_URL", "https://wp.example.com")
monkeypatch.setenv("WORDPRESS_SITE1_USERNAME", "admin")
monkeypatch.setenv("GITEA_SITE1_URL", "https://git.example.com")
monkeypatch.setenv("GITEA_SITE1_TOKEN", "tok_123")
manager = SiteManager()
count = manager.discover_sites(["wordpress", "gitea"])
assert count == 2
def test_reserved_words_skipped(self, monkeypatch):
"""Reserved words like LIMIT, RATE should not become site IDs."""
monkeypatch.setenv("WORDPRESS_LIMIT_PER_MINUTE", "100")
monkeypatch.setenv("WORDPRESS_RATE_PER_HOUR", "500")
manager = SiteManager()
count = manager.discover_sites(["wordpress"])
assert count == 0
def test_incomplete_config_skipped(self, monkeypatch):
"""Sites with no config data (only prefix match) should be skipped."""
# WORDPRESS_SITE1_ exists as prefix but no actual config keys
# This is hard to test directly since env vars always have a value
# Instead, test that a site with only site_id and plugin_type (no config) is skipped
manager = SiteManager()
count = manager.discover_sites(["wordpress"])
assert count == 0
class TestSiteManagerCounts:
"""Test counting and listing methods."""
@pytest.fixture
def populated_manager(self):
manager = SiteManager()
for i in range(3):
config = SiteConfig(
site_id=f"site{i+1}",
plugin_type="wordpress",
url=f"https://wp{i+1}.example.com",
)
manager.register_site(config)
config = SiteConfig(
site_id="repo1",
plugin_type="gitea",
url="https://git.example.com",
)
manager.register_site(config)
return manager
def test_get_count(self, populated_manager):
assert populated_manager.get_count() == 4
def test_get_count_by_type(self, populated_manager):
counts = populated_manager.get_count_by_type()
assert counts["wordpress"] == 3
assert counts["gitea"] == 1
def test_get_sites_by_type(self, populated_manager):
wp_sites = populated_manager.get_sites_by_type("wordpress")
assert len(wp_sites) == 3
def test_get_sites_by_type_empty(self, populated_manager):
assert populated_manager.get_sites_by_type("nonexistent") == []
def test_list_all_sites(self, populated_manager):
all_sites = populated_manager.list_all_sites()
assert len(all_sites) == 4
plugin_types = {s["plugin_type"] for s in all_sites}
assert plugin_types == {"wordpress", "gitea"}
def test_repr(self, populated_manager):
r = repr(populated_manager)
assert "SiteManager" in r
assert "total=4" in r

183
tests/test_tool_registry.py Normal file
View File

@@ -0,0 +1,183 @@
"""Tests for Tool Registry (core/tool_registry.py)."""
import pytest
from core.tool_registry import ToolDefinition, ToolRegistry
async def _dummy_handler(**kwargs):
"""Dummy async handler for testing."""
return {"ok": True}
async def _another_handler(**kwargs):
return {"ok": True}
@pytest.fixture
def registry():
return ToolRegistry()
@pytest.fixture
def sample_tool():
return ToolDefinition(
name="wordpress_list_posts",
description="List WordPress posts",
handler=_dummy_handler,
plugin_type="wordpress",
required_scope="read",
)
class TestToolDefinition:
"""Test ToolDefinition model."""
def test_basic_creation(self):
tool = ToolDefinition(
name="wordpress_get_post",
description="Get a post",
handler=_dummy_handler,
plugin_type="wordpress",
)
assert tool.name == "wordpress_get_post"
assert tool.required_scope == "read" # default
def test_custom_scope(self):
tool = ToolDefinition(
name="wordpress_create_post",
description="Create a post",
handler=_dummy_handler,
plugin_type="wordpress",
required_scope="write",
)
assert tool.required_scope == "write"
def test_default_input_schema(self):
tool = ToolDefinition(
name="test_tool",
description="Test",
handler=_dummy_handler,
plugin_type="test",
)
assert tool.input_schema == {"type": "object", "properties": {}}
def test_custom_input_schema(self):
schema = {
"type": "object",
"properties": {"site": {"type": "string"}},
"required": ["site"],
}
tool = ToolDefinition(
name="test_tool",
description="Test",
handler=_dummy_handler,
plugin_type="test",
input_schema=schema,
)
assert tool.input_schema["required"] == ["site"]
class TestToolRegistration:
"""Test tool registration."""
def test_register_single(self, registry, sample_tool):
registry.register(sample_tool)
assert registry.get_count() == 1
def test_duplicate_name_raises(self, registry, sample_tool):
registry.register(sample_tool)
with pytest.raises(ValueError, match="already registered"):
registry.register(sample_tool)
def test_register_many(self, registry):
tools = [
ToolDefinition(
name=f"tool_{i}",
description=f"Tool {i}",
handler=_dummy_handler,
plugin_type="test",
)
for i in range(5)
]
count = registry.register_many(tools)
assert count == 5
assert registry.get_count() == 5
def test_register_many_skips_duplicates(self, registry, sample_tool):
registry.register(sample_tool)
tools = [
sample_tool, # duplicate
ToolDefinition(
name="wordpress_create_post",
description="Create post",
handler=_dummy_handler,
plugin_type="wordpress",
),
]
count = registry.register_many(tools)
assert count == 1 # only the non-duplicate
assert registry.get_count() == 2
class TestToolRetrieval:
"""Test tool retrieval and filtering."""
@pytest.fixture
def populated_registry(self, registry):
tools = [
ToolDefinition(
name="wordpress_list_posts",
description="List posts",
handler=_dummy_handler,
plugin_type="wordpress",
),
ToolDefinition(
name="wordpress_create_post",
description="Create post",
handler=_dummy_handler,
plugin_type="wordpress",
required_scope="write",
),
ToolDefinition(
name="gitea_list_repos",
description="List repos",
handler=_another_handler,
plugin_type="gitea",
),
]
registry.register_many(tools)
return registry
def test_get_by_name(self, populated_registry):
tool = populated_registry.get_by_name("wordpress_list_posts")
assert tool is not None
assert tool.description == "List posts"
def test_get_by_name_not_found(self, populated_registry):
assert populated_registry.get_by_name("nonexistent") is None
def test_get_by_plugin_type(self, populated_registry):
wp_tools = populated_registry.get_by_plugin_type("wordpress")
assert len(wp_tools) == 2
def test_get_by_plugin_type_empty(self, populated_registry):
assert populated_registry.get_by_plugin_type("n8n") == []
def test_get_all(self, populated_registry):
all_tools = populated_registry.get_all()
assert len(all_tools) == 3
def test_get_count_by_plugin(self, populated_registry):
counts = populated_registry.get_count_by_plugin()
assert counts["wordpress"] == 2
assert counts["gitea"] == 1
def test_clear(self, populated_registry):
populated_registry.clear()
assert populated_registry.get_count() == 0
def test_repr(self, populated_registry):
r = repr(populated_registry)
assert "ToolRegistry" in r
assert "total=3" in r

View File

@@ -0,0 +1,303 @@
"""Tests for WooCommerce Plugin (plugins/woocommerce/).
Integration tests covering plugin initialization, configuration validation,
tool specifications, handler delegation, and health checks.
"""
from unittest.mock import AsyncMock
import pytest
from plugins.base import BasePlugin
from plugins.woocommerce.plugin import WooCommercePlugin
# --- WooCommercePlugin Initialization ---
class TestWooCommercePluginInit:
"""Test WooCommerce plugin initialization."""
VALID_CONFIG = {
"url": "https://shop.example.com",
"username": "admin",
"app_password": "xxxx xxxx xxxx xxxx",
}
def test_create_with_valid_config(self):
"""Should initialize with valid credentials."""
plugin = WooCommercePlugin(self.VALID_CONFIG)
assert plugin.project_id is not None
assert plugin.client is not None
assert isinstance(plugin, BasePlugin)
def test_handlers_initialized(self):
"""Should initialize all WooCommerce handlers."""
plugin = WooCommercePlugin(self.VALID_CONFIG)
assert plugin.products is not None
assert plugin.orders is not None
assert plugin.customers is not None
assert plugin.coupons is not None
assert plugin.reports is not None
def test_plugin_name(self):
"""Should return 'woocommerce' as plugin name."""
assert WooCommercePlugin.get_plugin_name() == "woocommerce"
def test_required_config_keys(self):
"""Should require url, username, app_password."""
keys = WooCommercePlugin.get_required_config_keys()
assert "url" in keys
assert "username" in keys
assert "app_password" in keys
def test_missing_url_raises(self):
"""Should raise ValueError for missing URL."""
config = {"username": "admin", "app_password": "xxxx"}
with pytest.raises(ValueError, match="Missing required configuration"):
WooCommercePlugin(config)
def test_missing_credentials_raises(self):
"""Should raise ValueError for missing credentials."""
config = {"url": "https://shop.example.com"}
with pytest.raises(ValueError, match="Missing required configuration"):
WooCommercePlugin(config)
def test_custom_project_id(self):
"""Should accept custom project_id."""
plugin = WooCommercePlugin(self.VALID_CONFIG, project_id="wc_myshop")
assert plugin.project_id == "wc_myshop"
def test_auto_generated_project_id(self):
"""Should auto-generate project_id from config."""
plugin = WooCommercePlugin(self.VALID_CONFIG)
assert plugin.project_id.startswith("woocommerce")
def test_uses_wordpress_client(self):
"""Should create a WordPressClient (shared API client)."""
from plugins.wordpress.client import WordPressClient
plugin = WooCommercePlugin(self.VALID_CONFIG)
assert isinstance(plugin.client, WordPressClient)
assert plugin.client.site_url == "https://shop.example.com"
# --- Tool Specifications ---
class TestWooCommerceToolSpecs:
"""Test WooCommerce tool specification generation."""
def test_specs_not_empty(self):
"""Should return non-empty tool specifications."""
specs = WooCommercePlugin.get_tool_specifications()
assert len(specs) > 0
def test_specs_count(self):
"""Should return exactly 28 tool specs."""
specs = WooCommercePlugin.get_tool_specifications()
assert len(specs) == 28
def test_specs_have_required_fields(self):
"""Each spec should have name, method_name, description, schema, scope."""
specs = WooCommercePlugin.get_tool_specifications()
for spec in specs:
assert "name" in spec, f"Missing 'name' in spec"
assert "method_name" in spec, f"Missing 'method_name' in {spec.get('name')}"
assert "description" in spec, f"Missing 'description' in {spec.get('name')}"
assert "schema" in spec, f"Missing 'schema' in {spec.get('name')}"
assert "scope" in spec, f"Missing 'scope' in {spec.get('name')}"
def test_specs_scope_values(self):
"""All scopes should be valid (read, write, admin)."""
specs = WooCommercePlugin.get_tool_specifications()
valid_scopes = {"read", "write", "admin"}
for spec in specs:
assert spec["scope"] in valid_scopes, f"Invalid scope '{spec['scope']}' in {spec['name']}"
def test_specs_unique_names(self):
"""All tool names should be unique."""
specs = WooCommercePlugin.get_tool_specifications()
names = [s["name"] for s in specs]
assert len(names) == len(set(names)), f"Duplicate names: {[n for n in names if names.count(n) > 1]}"
def test_product_tools_present(self):
"""Should include product management tools."""
specs = WooCommercePlugin.get_tool_specifications()
names = {s["name"] for s in specs}
expected = {"list_products", "get_product", "create_product", "update_product", "delete_product"}
assert expected.issubset(names), f"Missing product tools: {expected - names}"
def test_order_tools_present(self):
"""Should include order management tools."""
specs = WooCommercePlugin.get_tool_specifications()
names = {s["name"] for s in specs}
expected = {"list_orders", "get_order", "create_order", "update_order_status", "delete_order"}
assert expected.issubset(names), f"Missing order tools: {expected - names}"
def test_customer_tools_present(self):
"""Should include customer tools."""
specs = WooCommercePlugin.get_tool_specifications()
names = {s["name"] for s in specs}
assert "list_customers" in names
assert "create_customer" in names
def test_coupon_tools_present(self):
"""Should include coupon tools."""
specs = WooCommercePlugin.get_tool_specifications()
names = {s["name"] for s in specs}
assert "list_coupons" in names
assert "create_coupon" in names
def test_report_tools_present(self):
"""Should include report tools."""
specs = WooCommercePlugin.get_tool_specifications()
names = {s["name"] for s in specs}
assert "get_sales_report" in names
assert "get_top_sellers" in names
def test_no_wordpress_tools_leaked(self):
"""Should NOT include WordPress-core tools (posts, pages, etc)."""
specs = WooCommercePlugin.get_tool_specifications()
names = {s["name"] for s in specs}
wp_tools = {"list_posts", "create_post", "list_categories", "list_media"}
leaked = names & wp_tools
assert len(leaked) == 0, f"WordPress tools leaked into WooCommerce: {leaked}"
# --- Handler Delegation ---
class TestWooCommerceHandlerDelegation:
"""Test that plugin methods delegate to handlers correctly."""
VALID_CONFIG = {
"url": "https://shop.example.com",
"username": "admin",
"app_password": "xxxx xxxx xxxx xxxx",
}
@pytest.fixture
def plugin(self):
return WooCommercePlugin(self.VALID_CONFIG)
@pytest.mark.asyncio
async def test_list_products_delegates(self, plugin):
"""list_products should delegate to products handler."""
plugin.products.list_products = AsyncMock(return_value={"products": []})
result = await plugin.list_products(per_page=10)
plugin.products.list_products.assert_called_once_with(per_page=10)
assert result == {"products": []}
@pytest.mark.asyncio
async def test_create_product_delegates(self, plugin):
"""create_product should delegate to products handler."""
plugin.products.create_product = AsyncMock(return_value={"id": 99})
result = await plugin.create_product(name="Widget", regular_price="19.99")
plugin.products.create_product.assert_called_once_with(name="Widget", regular_price="19.99")
@pytest.mark.asyncio
async def test_list_orders_delegates(self, plugin):
"""list_orders should delegate to orders handler."""
plugin.orders.list_orders = AsyncMock(return_value=[])
await plugin.list_orders(status="processing")
plugin.orders.list_orders.assert_called_once_with(status="processing")
@pytest.mark.asyncio
async def test_get_order_delegates(self, plugin):
"""get_order should delegate to orders handler."""
plugin.orders.get_order = AsyncMock(return_value={"id": 1, "status": "completed"})
result = await plugin.get_order(order_id=1)
plugin.orders.get_order.assert_called_once_with(order_id=1)
@pytest.mark.asyncio
async def test_list_customers_delegates(self, plugin):
"""list_customers should delegate to customers handler."""
plugin.customers.list_customers = AsyncMock(return_value=[])
await plugin.list_customers()
plugin.customers.list_customers.assert_called_once()
@pytest.mark.asyncio
async def test_create_coupon_delegates(self, plugin):
"""create_coupon should delegate to coupons handler."""
plugin.coupons.create_coupon = AsyncMock(return_value={"id": 5})
result = await plugin.create_coupon(code="SAVE10", discount_type="percent")
plugin.coupons.create_coupon.assert_called_once_with(code="SAVE10", discount_type="percent")
@pytest.mark.asyncio
async def test_get_sales_report_delegates(self, plugin):
"""get_sales_report should delegate to reports handler."""
plugin.reports.get_sales_report = AsyncMock(return_value={"total": 1000})
result = await plugin.get_sales_report(period="month")
plugin.reports.get_sales_report.assert_called_once_with(period="month")
# --- Health Check ---
class TestWooCommerceHealthCheck:
"""Test WooCommerce health check."""
VALID_CONFIG = {
"url": "https://shop.example.com",
"username": "admin",
"app_password": "xxxx xxxx xxxx xxxx",
}
@pytest.fixture
def plugin(self):
return WooCommercePlugin(self.VALID_CONFIG)
@pytest.mark.asyncio
async def test_healthy_when_wc_available(self, plugin):
"""Should report healthy when WooCommerce is available."""
plugin.client.check_woocommerce = AsyncMock(
return_value={"available": True, "version": "8.5.0"}
)
result = await plugin.health_check()
assert result["healthy"] is True
assert result["version"] == "8.5.0"
assert result["plugin_type"] == "woocommerce"
@pytest.mark.asyncio
async def test_unhealthy_when_wc_unavailable(self, plugin):
"""Should report unhealthy when WooCommerce is not available."""
plugin.client.check_woocommerce = AsyncMock(
return_value={"available": False, "version": None}
)
result = await plugin.health_check()
assert result["healthy"] is False
assert "not available" in result["message"]
@pytest.mark.asyncio
async def test_unhealthy_on_exception(self, plugin):
"""Should report unhealthy on network errors."""
plugin.client.check_woocommerce = AsyncMock(side_effect=Exception("Connection refused"))
result = await plugin.health_check()
assert result["healthy"] is False
assert "Connection refused" in result["message"]
# --- Plugin Info ---
class TestWooCommercePluginInfo:
"""Test plugin info methods."""
VALID_CONFIG = {
"url": "https://shop.example.com",
"username": "admin",
"app_password": "xxxx xxxx xxxx xxxx",
}
def test_get_project_info(self):
"""Should return structured project info."""
plugin = WooCommercePlugin(self.VALID_CONFIG, project_id="wc_shop1")
info = plugin.get_project_info()
assert info["project_id"] == "wc_shop1"
assert info["plugin_type"] == "woocommerce"
def test_legacy_get_tools_empty(self):
"""Legacy get_tools should return empty list."""
plugin = WooCommercePlugin(self.VALID_CONFIG)
assert plugin.get_tools() == []

View File

@@ -0,0 +1,535 @@
"""Tests for WordPress Plugin (plugins/wordpress/).
Integration tests covering plugin initialization, configuration validation,
tool specifications, handler delegation, client behavior, and health checks.
"""
import base64
import json
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from plugins.base import BasePlugin, PluginRegistry
from plugins.wordpress.client import AuthenticationError, ConfigurationError, WordPressClient
from plugins.wordpress.plugin import WordPressPlugin
# --- WordPressClient Tests ---
class TestWordPressClientInit:
"""Test WordPressClient initialization and validation."""
def test_valid_initialization(self):
"""Should initialize with valid credentials."""
client = WordPressClient(
site_url="https://example.com",
username="admin",
app_password="xxxx xxxx xxxx xxxx",
)
assert client.site_url == "https://example.com"
assert client.api_base == "https://example.com/wp-json/wp/v2"
assert client.wc_api_base == "https://example.com/wp-json/wc/v3"
assert client.username == "admin"
def test_trailing_slash_stripped(self):
"""Should strip trailing slash from site URL."""
client = WordPressClient(
site_url="https://example.com/",
username="admin",
app_password="xxxx",
)
assert client.site_url == "https://example.com"
assert client.api_base == "https://example.com/wp-json/wp/v2"
def test_auth_header_created(self):
"""Should create proper Basic auth header."""
client = WordPressClient(
site_url="https://example.com",
username="admin",
app_password="secret123",
)
expected_token = base64.b64encode(b"admin:secret123").decode()
assert client.auth_header == f"Basic {expected_token}"
def test_missing_url_raises(self):
"""Should raise ConfigurationError for empty URL."""
with pytest.raises(ConfigurationError, match="Site URL is not configured"):
WordPressClient(site_url="", username="admin", app_password="xxxx")
def test_missing_username_raises(self):
"""Should raise ConfigurationError for empty username."""
with pytest.raises(ConfigurationError, match="Username is not configured"):
WordPressClient(site_url="https://example.com", username="", app_password="xxxx")
def test_missing_password_raises(self):
"""Should raise ConfigurationError for empty app password."""
with pytest.raises(ConfigurationError, match="App password is not configured"):
WordPressClient(site_url="https://example.com", username="admin", app_password="")
def test_none_url_raises(self):
"""Should raise ConfigurationError for None URL."""
with pytest.raises(ConfigurationError):
WordPressClient(site_url=None, username="admin", app_password="xxxx")
class TestWordPressClientErrorParsing:
"""Test error response parsing."""
@pytest.fixture
def client(self):
return WordPressClient(
site_url="https://example.com",
username="admin",
app_password="xxxx",
)
def test_parse_json_error(self, client):
"""Should parse JSON error responses."""
error_text = json.dumps({"code": "rest_forbidden", "message": "Sorry, you are not allowed."})
result = client._parse_error_response(403, error_text)
assert result["error_code"] == "ACCESS_DENIED"
assert result["status_code"] == 403
assert result["wp_error_code"] == "rest_forbidden"
def test_parse_non_json_error(self, client):
"""Should handle non-JSON error responses."""
result = client._parse_error_response(500, "Internal Server Error")
assert result["error_code"] == "SERVER_ERROR"
assert result["wp_error_code"] == "unknown_error"
def test_parse_auth_error(self, client):
"""Should provide auth-specific error messages."""
error_text = json.dumps({"code": "invalid_auth", "message": "Bad credentials"})
result = client._parse_error_response(401, error_text)
assert result["error_code"] == "AUTH_FAILED"
assert "Authentication failed" in result["message"]
assert "Application Password" in result["message"]
def test_parse_woocommerce_auth_error(self, client):
"""Should provide WooCommerce-specific auth error."""
error_text = json.dumps({"code": "wc_auth", "message": "No permission"})
result = client._parse_error_response(401, error_text, use_woocommerce=True)
assert "WooCommerce" in result["message"]
assert "manage_woocommerce" in result["message"]
def test_parse_404_error(self, client):
"""Should parse 404 errors correctly."""
error_text = json.dumps({"code": "rest_no_route", "message": "No route found"})
result = client._parse_error_response(404, error_text)
assert result["error_code"] == "NOT_FOUND"
assert "not found" in result["message"].lower()
def test_parse_400_error_with_hints(self, client):
"""Should provide parameter hints for 400 errors."""
error_text = json.dumps({"code": "rest_invalid_param", "message": "Invalid param"})
result = client._parse_error_response(400, error_text)
assert result["error_code"] == "BAD_REQUEST"
assert "Hints" in result["message"]
def test_raw_response_truncated(self, client):
"""Should truncate raw response to 500 chars."""
error_text = "x" * 1000
result = client._parse_error_response(500, error_text)
assert len(result["raw_response"]) == 500
def test_unknown_status_code(self, client):
"""Should handle unknown HTTP status codes."""
result = client._parse_error_response(418, "I'm a teapot")
assert result["error_code"] == "HTTP_418"
class TestWordPressClientRequest:
"""Test HTTP request methods."""
@pytest.fixture
def client(self):
return WordPressClient(
site_url="https://example.com",
username="admin",
app_password="xxxx",
)
@pytest.mark.asyncio
async def test_request_filters_none_params(self, client):
"""Should filter None and empty values from params."""
mock_response = AsyncMock()
mock_response.status = 200
mock_response.json = AsyncMock(return_value={"id": 1})
mock_session = AsyncMock()
mock_session.request = MagicMock(return_value=AsyncMock(
__aenter__=AsyncMock(return_value=mock_response),
__aexit__=AsyncMock(return_value=False),
))
with patch("aiohttp.ClientSession") as mock_cls:
mock_cls.return_value = AsyncMock(
__aenter__=AsyncMock(return_value=mock_session),
__aexit__=AsyncMock(return_value=False),
)
result = await client.request(
"GET", "posts",
params={"status": "publish", "search": None, "tags": "", "ids": []},
)
assert result == {"id": 1}
# Verify filtered params
call_kwargs = mock_session.request.call_args
filtered_params = call_kwargs.kwargs.get("params") or call_kwargs[1].get("params", {})
if filtered_params:
assert "search" not in filtered_params
assert "tags" not in filtered_params
assert "ids" not in filtered_params
@pytest.mark.asyncio
async def test_request_raises_on_401(self, client):
"""Should raise AuthenticationError on 401."""
mock_response = AsyncMock()
mock_response.status = 401
mock_response.text = AsyncMock(
return_value=json.dumps({"code": "invalid_auth", "message": "Bad creds"})
)
mock_session = AsyncMock()
mock_session.request = MagicMock(return_value=AsyncMock(
__aenter__=AsyncMock(return_value=mock_response),
__aexit__=AsyncMock(return_value=False),
))
with patch("aiohttp.ClientSession") as mock_cls:
mock_cls.return_value = AsyncMock(
__aenter__=AsyncMock(return_value=mock_session),
__aexit__=AsyncMock(return_value=False),
)
with pytest.raises(AuthenticationError):
await client.request("GET", "posts")
@pytest.mark.asyncio
async def test_request_raises_on_500(self, client):
"""Should raise Exception on 500."""
mock_response = AsyncMock()
mock_response.status = 500
mock_response.text = AsyncMock(return_value="Internal Server Error")
mock_session = AsyncMock()
mock_session.request = MagicMock(return_value=AsyncMock(
__aenter__=AsyncMock(return_value=mock_response),
__aexit__=AsyncMock(return_value=False),
))
with patch("aiohttp.ClientSession") as mock_cls:
mock_cls.return_value = AsyncMock(
__aenter__=AsyncMock(return_value=mock_session),
__aexit__=AsyncMock(return_value=False),
)
with pytest.raises(Exception, match="SERVER_ERROR"):
await client.request("GET", "posts")
@pytest.mark.asyncio
async def test_get_convenience_method(self, client):
"""GET method should delegate to request."""
client.request = AsyncMock(return_value={"posts": []})
result = await client.get("posts", params={"per_page": 10})
client.request.assert_called_once_with(
"GET", "posts", params={"per_page": 10},
use_custom_namespace=False, use_woocommerce=False,
)
assert result == {"posts": []}
@pytest.mark.asyncio
async def test_post_convenience_method(self, client):
"""POST method should delegate to request."""
client.request = AsyncMock(return_value={"id": 1})
result = await client.post("posts", json_data={"title": "Test"})
client.request.assert_called_once()
assert result == {"id": 1}
@pytest.mark.asyncio
async def test_delete_convenience_method(self, client):
"""DELETE method should delegate to request."""
client.request = AsyncMock(return_value={"deleted": True})
result = await client.delete("posts/1", params={"force": True})
client.request.assert_called_once()
assert result == {"deleted": True}
@pytest.mark.asyncio
async def test_woocommerce_url(self, client):
"""WooCommerce requests should use wc/v3 base."""
client.request = AsyncMock(return_value={"available": True})
await client.get("products", use_woocommerce=True)
client.request.assert_called_once_with(
"GET", "products", params=None,
use_custom_namespace=False, use_woocommerce=True,
)
# --- WordPressPlugin Tests ---
class TestWordPressPluginInit:
"""Test WordPress plugin initialization."""
VALID_CONFIG = {
"url": "https://example.com",
"username": "admin",
"app_password": "xxxx xxxx xxxx xxxx",
}
def test_create_with_valid_config(self):
"""Should initialize with all required config keys."""
plugin = WordPressPlugin(self.VALID_CONFIG)
assert plugin.project_id is not None
assert plugin.client is not None
assert isinstance(plugin, BasePlugin)
def test_handlers_initialized(self):
"""Should initialize all core handlers."""
plugin = WordPressPlugin(self.VALID_CONFIG)
assert plugin.posts is not None
assert plugin.media is not None
assert plugin.taxonomy is not None
assert plugin.comments is not None
assert plugin.users is not None
assert plugin.site is not None
assert plugin.seo is not None
assert plugin.menus is not None
def test_wp_cli_none_without_container(self):
"""WP-CLI handler should be None without container config."""
plugin = WordPressPlugin(self.VALID_CONFIG)
assert plugin.wp_cli is None
def test_missing_url_raises(self):
"""Should raise ValueError for missing URL."""
config = {"username": "admin", "app_password": "xxxx"}
with pytest.raises(ValueError, match="Missing required configuration"):
WordPressPlugin(config)
def test_missing_username_raises(self):
"""Should raise ValueError for missing username."""
config = {"url": "https://example.com", "app_password": "xxxx"}
with pytest.raises(ValueError, match="Missing required configuration"):
WordPressPlugin(config)
def test_missing_password_raises(self):
"""Should raise ValueError for missing app_password."""
config = {"url": "https://example.com", "username": "admin"}
with pytest.raises(ValueError, match="Missing required configuration"):
WordPressPlugin(config)
def test_custom_project_id(self):
"""Should accept custom project_id."""
plugin = WordPressPlugin(self.VALID_CONFIG, project_id="wp_myblog")
assert plugin.project_id == "wp_myblog"
def test_auto_generated_project_id(self):
"""Should auto-generate project_id from config."""
plugin = WordPressPlugin(self.VALID_CONFIG)
assert plugin.project_id.startswith("wordpress")
def test_plugin_name(self):
"""Should return 'wordpress' as plugin name."""
assert WordPressPlugin.get_plugin_name() == "wordpress"
def test_required_config_keys(self):
"""Should require url, username, app_password."""
keys = WordPressPlugin.get_required_config_keys()
assert "url" in keys
assert "username" in keys
assert "app_password" in keys
class TestWordPressToolSpecifications:
"""Test tool specification generation."""
def test_specs_not_empty(self):
"""Should return non-empty tool specifications."""
specs = WordPressPlugin.get_tool_specifications()
assert len(specs) > 0
def test_specs_count(self):
"""Should return at least 65 tool specs (65 documented + possible additions)."""
specs = WordPressPlugin.get_tool_specifications()
assert len(specs) >= 65
def test_specs_have_required_fields(self):
"""Each spec should have name, method_name, description, schema, scope."""
specs = WordPressPlugin.get_tool_specifications()
for spec in specs:
assert "name" in spec, f"Missing 'name' in spec"
assert "method_name" in spec, f"Missing 'method_name' in {spec.get('name')}"
assert "description" in spec, f"Missing 'description' in {spec.get('name')}"
assert "schema" in spec, f"Missing 'schema' in {spec.get('name')}"
assert "scope" in spec, f"Missing 'scope' in {spec.get('name')}"
def test_specs_scope_values(self):
"""All scopes should be valid (read, write, admin)."""
specs = WordPressPlugin.get_tool_specifications()
valid_scopes = {"read", "write", "admin"}
for spec in specs:
assert spec["scope"] in valid_scopes, f"Invalid scope '{spec['scope']}' in {spec['name']}"
def test_specs_unique_names(self):
"""All tool names should be unique."""
specs = WordPressPlugin.get_tool_specifications()
names = [s["name"] for s in specs]
assert len(names) == len(set(names)), f"Duplicate tool names found: {[n for n in names if names.count(n) > 1]}"
def test_core_tools_present(self):
"""Should include key WordPress tools."""
specs = WordPressPlugin.get_tool_specifications()
names = {s["name"] for s in specs}
expected = {"list_posts", "create_post", "get_post", "update_post", "delete_post"}
assert expected.issubset(names), f"Missing core tools: {expected - names}"
def test_media_tools_present(self):
"""Should include media tools."""
specs = WordPressPlugin.get_tool_specifications()
names = {s["name"] for s in specs}
assert "list_media" in names
assert "upload_media_from_url" in names
def test_taxonomy_tools_present(self):
"""Should include taxonomy tools."""
specs = WordPressPlugin.get_tool_specifications()
names = {s["name"] for s in specs}
assert "list_categories" in names
assert "list_tags" in names
def test_wp_cli_tools_present(self):
"""Should include WP-CLI tools."""
specs = WordPressPlugin.get_tool_specifications()
names = {s["name"] for s in specs}
assert "wp_cache_flush" in names
assert "wp_db_check" in names
class TestWordPressHandlerDelegation:
"""Test that plugin methods delegate to handlers correctly."""
VALID_CONFIG = {
"url": "https://example.com",
"username": "admin",
"app_password": "xxxx xxxx xxxx xxxx",
}
@pytest.fixture
def plugin(self):
return WordPressPlugin(self.VALID_CONFIG)
@pytest.mark.asyncio
async def test_list_posts_delegates(self, plugin):
"""list_posts should delegate to posts handler."""
plugin.posts.list_posts = AsyncMock(return_value={"posts": []})
result = await plugin.list_posts(per_page=5)
plugin.posts.list_posts.assert_called_once_with(per_page=5)
assert result == {"posts": []}
@pytest.mark.asyncio
async def test_create_post_delegates(self, plugin):
"""create_post should delegate to posts handler."""
plugin.posts.create_post = AsyncMock(return_value={"id": 42})
result = await plugin.create_post(title="Test", content="Hello")
plugin.posts.create_post.assert_called_once_with(title="Test", content="Hello")
assert result == {"id": 42}
@pytest.mark.asyncio
async def test_list_media_delegates(self, plugin):
"""list_media should delegate to media handler."""
plugin.media.list_media = AsyncMock(return_value=[])
result = await plugin.list_media()
plugin.media.list_media.assert_called_once()
@pytest.mark.asyncio
async def test_list_categories_delegates(self, plugin):
"""list_categories should delegate to taxonomy handler."""
plugin.taxonomy.list_categories = AsyncMock(return_value=[])
result = await plugin.list_categories()
plugin.taxonomy.list_categories.assert_called_once()
@pytest.mark.asyncio
async def test_list_comments_delegates(self, plugin):
"""list_comments should delegate to comments handler."""
plugin.comments.list_comments = AsyncMock(return_value=[])
result = await plugin.list_comments()
plugin.comments.list_comments.assert_called_once()
@pytest.mark.asyncio
async def test_wp_cli_not_available(self, plugin):
"""WP-CLI methods should return error when not configured."""
result = await plugin.wp_cache_flush()
assert "error" in result.lower() or "error" in json.loads(result)
@pytest.mark.asyncio
async def test_health_check_delegates(self, plugin):
"""health_check should delegate to site handler."""
plugin.site.health_check = AsyncMock(return_value={"healthy": True})
result = await plugin.health_check()
assert result["healthy"] is True
class TestWordPressPluginInfo:
"""Test plugin info and metadata methods."""
VALID_CONFIG = {
"url": "https://example.com",
"username": "admin",
"app_password": "xxxx xxxx xxxx xxxx",
}
def test_get_project_info(self):
"""Should return structured project info."""
plugin = WordPressPlugin(self.VALID_CONFIG, project_id="wp_test")
info = plugin.get_project_info()
assert info["project_id"] == "wp_test"
assert info["plugin_type"] == "wordpress"
assert "url" in info["config_keys"]
def test_get_tools_returns_empty(self):
"""Legacy get_tools should return empty list (Option B architecture)."""
plugin = WordPressPlugin(self.VALID_CONFIG)
assert plugin.get_tools() == []
# --- PluginRegistry Tests ---
class TestPluginRegistryWithWordPress:
"""Test PluginRegistry with WordPress plugin."""
def test_register_wordpress(self):
"""Should register WordPress plugin type."""
reg = PluginRegistry()
reg.register("wordpress", WordPressPlugin)
assert reg.is_registered("wordpress")
assert "wordpress" in reg.get_registered_types()
def test_create_instance(self):
"""Should create WordPress plugin instance."""
reg = PluginRegistry()
reg.register("wordpress", WordPressPlugin)
config = {
"url": "https://example.com",
"username": "admin",
"app_password": "xxxx",
}
instance = reg.create_instance("wordpress", "wp_site1", config)
assert isinstance(instance, WordPressPlugin)
assert instance.project_id == "wp_site1"
def test_register_non_baseplugin_raises(self):
"""Should reject classes not inheriting BasePlugin."""
reg = PluginRegistry()
with pytest.raises(TypeError, match="must inherit from BasePlugin"):
reg.register("bad", dict)
def test_unknown_type_raises(self):
"""Should raise KeyError for unknown plugin type."""
reg = PluginRegistry()
with pytest.raises(KeyError, match="Unknown plugin type"):
reg.create_instance("nonexistent", "id1", {})