mirror of
https://github.com/latinogino/dolibarr-mcp.git
synced 2026-04-29 21:15:35 +02:00
Add comprehensive test suite for all Dolibarr MCP operations
This commit is contained in:
latinogino
@@ -1,247 +1,375 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Test script for Dolibarr MCP Server
|
||||
Tests all CRUD operations and API connectivity
|
||||
"""
|
||||
"""Comprehensive test suite for Dolibarr MCP Server."""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
from dotenv import load_dotenv
|
||||
|
||||
# Add parent directory to path for imports
|
||||
sys.path.insert(0, str(Path(__file__).parent))
|
||||
# Add parent directory to path
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from src.dolibarr_mcp.config import Config
|
||||
from src.dolibarr_mcp.dolibarr_client import DolibarrClient, DolibarrAPIError
|
||||
|
||||
# Load environment variables
|
||||
load_dotenv()
|
||||
|
||||
|
||||
class TestColors:
|
||||
"""ANSI color codes for terminal output."""
|
||||
HEADER = '\033[95m'
|
||||
OKBLUE = '\033[94m'
|
||||
OKCYAN = '\033[96m'
|
||||
OKGREEN = '\033[92m'
|
||||
WARNING = '\033[93m'
|
||||
FAIL = '\033[91m'
|
||||
ENDC = '\033[0m'
|
||||
BOLD = '\033[1m'
|
||||
UNDERLINE = '\033[4m'
|
||||
|
||||
|
||||
def print_header(text):
|
||||
"""Print a formatted header."""
|
||||
print(f"\n{TestColors.HEADER}{TestColors.BOLD}{'=' * 60}{TestColors.ENDC}")
|
||||
print(f"{TestColors.HEADER}{TestColors.BOLD}{text}{TestColors.ENDC}")
|
||||
print(f"{TestColors.HEADER}{TestColors.BOLD}{'=' * 60}{TestColors.ENDC}")
|
||||
|
||||
|
||||
def print_test(name, result, details=""):
|
||||
"""Print test result."""
|
||||
if result:
|
||||
status = f"{TestColors.OKGREEN}✅ PASS{TestColors.ENDC}"
|
||||
else:
|
||||
status = f"{TestColors.FAIL}❌ FAIL{TestColors.ENDC}"
|
||||
|
||||
print(f" {status} - {name}")
|
||||
if details:
|
||||
print(f" {TestColors.OKCYAN}{details}{TestColors.ENDC}")
|
||||
|
||||
|
||||
async def test_connection(client):
|
||||
"""Test basic API connection."""
|
||||
print("\n1. Testing API Connection...")
|
||||
print_header("🔌 CONNECTION TEST")
|
||||
|
||||
try:
|
||||
result = await client.get_status()
|
||||
print("✅ API Connection successful")
|
||||
print(f" Response: {json.dumps(result, indent=2)[:200]}...")
|
||||
print_test("API Connection", True, f"Dolibarr v{result.get('dolibarr_version', 'unknown')}")
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"❌ Connection failed: {e}")
|
||||
print_test("API Connection", False, str(e))
|
||||
return False
|
||||
|
||||
|
||||
async def test_customer_crud(client):
|
||||
"""Test Customer CRUD operations."""
|
||||
print("\n2. Testing Customer CRUD Operations...")
|
||||
async def test_customers(client):
|
||||
"""Test customer CRUD operations."""
|
||||
print_header("👥 CUSTOMER MANAGEMENT")
|
||||
|
||||
test_customer_id = None
|
||||
|
||||
try:
|
||||
# CREATE
|
||||
print(" Creating customer...")
|
||||
customer = await client.create_customer(
|
||||
name="Test Company MCP",
|
||||
email="test@mcp-example.com",
|
||||
# List customers
|
||||
customers = await client.get_customers(limit=5)
|
||||
print_test("List Customers", True, f"Found {len(customers)} customers")
|
||||
|
||||
# Create customer
|
||||
new_customer = await client.create_customer(
|
||||
name=f"Test Customer {datetime.now().strftime('%Y%m%d%H%M%S')}",
|
||||
email="test@example.com",
|
||||
phone="+1234567890",
|
||||
address="123 Test Street",
|
||||
town="Test City",
|
||||
zip="12345"
|
||||
)
|
||||
customer_id = customer.get('id') or customer.get('rowid')
|
||||
print(f" ✅ Created customer ID: {customer_id}")
|
||||
test_customer_id = new_customer if isinstance(new_customer, int) else new_customer.get('id')
|
||||
print_test("Create Customer", True, f"Created ID: {test_customer_id}")
|
||||
|
||||
# READ
|
||||
print(" Reading customer...")
|
||||
retrieved = await client.get_customer_by_id(customer_id)
|
||||
print(f" ✅ Retrieved customer: {retrieved.get('name')}")
|
||||
|
||||
# UPDATE
|
||||
print(" Updating customer...")
|
||||
updated = await client.update_customer(
|
||||
customer_id,
|
||||
email="updated@mcp-example.com"
|
||||
)
|
||||
print(f" ✅ Updated customer")
|
||||
|
||||
# DELETE
|
||||
print(" Deleting customer...")
|
||||
await client.delete_customer(customer_id)
|
||||
print(f" ✅ Deleted customer")
|
||||
# Get customer by ID
|
||||
if test_customer_id:
|
||||
customer = await client.get_customer_by_id(test_customer_id)
|
||||
print_test("Get Customer by ID", True, f"Retrieved: {customer.get('name', 'Unknown')}")
|
||||
|
||||
# Update customer
|
||||
updated = await client.update_customer(
|
||||
test_customer_id,
|
||||
email="updated@example.com"
|
||||
)
|
||||
print_test("Update Customer", True, "Email updated")
|
||||
|
||||
# Delete customer
|
||||
deleted = await client.delete_customer(test_customer_id)
|
||||
print_test("Delete Customer", True, f"Deleted ID: {test_customer_id}")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f" ❌ Customer CRUD failed: {e}")
|
||||
print_test("Customer Operations", False, str(e))
|
||||
# Try to clean up
|
||||
if test_customer_id:
|
||||
try:
|
||||
await client.delete_customer(test_customer_id)
|
||||
except:
|
||||
pass
|
||||
return False
|
||||
|
||||
|
||||
async def test_product_crud(client):
|
||||
"""Test Product CRUD operations."""
|
||||
print("\n3. Testing Product CRUD Operations...")
|
||||
async def test_products(client):
|
||||
"""Test product CRUD operations."""
|
||||
print_header("📦 PRODUCT MANAGEMENT")
|
||||
|
||||
test_product_id = None
|
||||
|
||||
try:
|
||||
# CREATE
|
||||
print(" Creating product...")
|
||||
product = await client.create_product(
|
||||
label="Test Product MCP",
|
||||
# List products
|
||||
products = await client.get_products(limit=5)
|
||||
print_test("List Products", True, f"Found {len(products)} products")
|
||||
|
||||
# Create product
|
||||
new_product = await client.create_product(
|
||||
label=f"Test Product {datetime.now().strftime('%Y%m%d%H%M%S')}",
|
||||
price=99.99,
|
||||
description="A test product created via MCP",
|
||||
description="Test product description",
|
||||
stock=100
|
||||
)
|
||||
product_id = product.get('id') or product.get('rowid')
|
||||
print(f" ✅ Created product ID: {product_id}")
|
||||
test_product_id = new_product if isinstance(new_product, int) else new_product.get('id')
|
||||
print_test("Create Product", True, f"Created ID: {test_product_id}")
|
||||
|
||||
# READ
|
||||
print(" Reading product...")
|
||||
retrieved = await client.get_product_by_id(product_id)
|
||||
print(f" ✅ Retrieved product: {retrieved.get('label')}")
|
||||
|
||||
# UPDATE
|
||||
print(" Updating product...")
|
||||
updated = await client.update_product(
|
||||
product_id,
|
||||
price=149.99
|
||||
)
|
||||
print(f" ✅ Updated product price")
|
||||
|
||||
# DELETE
|
||||
print(" Deleting product...")
|
||||
await client.delete_product(product_id)
|
||||
print(f" ✅ Deleted product")
|
||||
# Get product by ID
|
||||
if test_product_id:
|
||||
product = await client.get_product_by_id(test_product_id)
|
||||
print_test("Get Product by ID", True, f"Retrieved: {product.get('label', 'Unknown')}")
|
||||
|
||||
# Update product
|
||||
updated = await client.update_product(
|
||||
test_product_id,
|
||||
price=149.99
|
||||
)
|
||||
print_test("Update Product", True, "Price updated")
|
||||
|
||||
# Delete product
|
||||
deleted = await client.delete_product(test_product_id)
|
||||
print_test("Delete Product", True, f"Deleted ID: {test_product_id}")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f" ❌ Product CRUD failed: {e}")
|
||||
print_test("Product Operations", False, str(e))
|
||||
# Try to clean up
|
||||
if test_product_id:
|
||||
try:
|
||||
await client.delete_product(test_product_id)
|
||||
except:
|
||||
pass
|
||||
return False
|
||||
|
||||
|
||||
async def test_list_operations(client):
|
||||
"""Test listing operations."""
|
||||
print("\n4. Testing List Operations...")
|
||||
async def test_users(client):
|
||||
"""Test user operations."""
|
||||
print_header("👤 USER MANAGEMENT")
|
||||
|
||||
try:
|
||||
# List users
|
||||
print(" Listing users...")
|
||||
users = await client.get_users(limit=5)
|
||||
print(f" ✅ Found {len(users)} users")
|
||||
print_test("List Users", True, f"Found {len(users)} users")
|
||||
|
||||
# List customers
|
||||
print(" Listing customers...")
|
||||
customers = await client.get_customers(limit=5)
|
||||
print(f" ✅ Found {len(customers)} customers")
|
||||
|
||||
# List products
|
||||
print(" Listing products...")
|
||||
products = await client.get_products(limit=5)
|
||||
print(f" ✅ Found {len(products)} products")
|
||||
|
||||
# List invoices
|
||||
print(" Listing invoices...")
|
||||
invoices = await client.get_invoices(limit=5)
|
||||
print(f" ✅ Found {len(invoices)} invoices")
|
||||
|
||||
# List orders
|
||||
print(" Listing orders...")
|
||||
orders = await client.get_orders(limit=5)
|
||||
print(f" ✅ Found {len(orders)} orders")
|
||||
|
||||
# List contacts
|
||||
print(" Listing contacts...")
|
||||
contacts = await client.get_contacts(limit=5)
|
||||
print(f" ✅ Found {len(contacts)} contacts")
|
||||
if users:
|
||||
# Get first user details
|
||||
first_user = users[0]
|
||||
user_id = first_user.get('id')
|
||||
if user_id:
|
||||
user = await client.get_user_by_id(user_id)
|
||||
print_test("Get User by ID", True, f"Retrieved: {user.get('login', 'Unknown')}")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f" ❌ List operations failed: {e}")
|
||||
print_test("User Operations", False, str(e))
|
||||
return False
|
||||
|
||||
|
||||
async def test_invoices(client):
|
||||
"""Test invoice operations."""
|
||||
print_header("📄 INVOICE MANAGEMENT")
|
||||
|
||||
try:
|
||||
# List invoices
|
||||
invoices = await client.get_invoices(limit=5)
|
||||
print_test("List Invoices", True, f"Found {len(invoices)} invoices")
|
||||
|
||||
if invoices:
|
||||
# Get first invoice details
|
||||
first_invoice = invoices[0]
|
||||
invoice_id = first_invoice.get('id')
|
||||
if invoice_id:
|
||||
invoice = await client.get_invoice_by_id(invoice_id)
|
||||
print_test("Get Invoice by ID", True, f"Retrieved Invoice #{invoice.get('ref', 'Unknown')}")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print_test("Invoice Operations", False, str(e))
|
||||
return False
|
||||
|
||||
|
||||
async def test_orders(client):
|
||||
"""Test order operations."""
|
||||
print_header("📋 ORDER MANAGEMENT")
|
||||
|
||||
try:
|
||||
# List orders
|
||||
orders = await client.get_orders(limit=5)
|
||||
print_test("List Orders", True, f"Found {len(orders)} orders")
|
||||
|
||||
if orders:
|
||||
# Get first order details
|
||||
first_order = orders[0]
|
||||
order_id = first_order.get('id')
|
||||
if order_id:
|
||||
order = await client.get_order_by_id(order_id)
|
||||
print_test("Get Order by ID", True, f"Retrieved Order #{order.get('ref', 'Unknown')}")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print_test("Order Operations", False, str(e))
|
||||
return False
|
||||
|
||||
|
||||
async def test_contacts(client):
|
||||
"""Test contact operations."""
|
||||
print_header("📇 CONTACT MANAGEMENT")
|
||||
|
||||
test_contact_id = None
|
||||
|
||||
try:
|
||||
# List contacts
|
||||
contacts = await client.get_contacts(limit=5)
|
||||
print_test("List Contacts", True, f"Found {len(contacts)} contacts")
|
||||
|
||||
# Create contact
|
||||
new_contact = await client.create_contact(
|
||||
firstname="Test",
|
||||
lastname=f"Contact {datetime.now().strftime('%Y%m%d%H%M%S')}",
|
||||
email="testcontact@example.com"
|
||||
)
|
||||
test_contact_id = new_contact if isinstance(new_contact, int) else new_contact.get('id')
|
||||
print_test("Create Contact", True, f"Created ID: {test_contact_id}")
|
||||
|
||||
# Get contact by ID
|
||||
if test_contact_id:
|
||||
contact = await client.get_contact_by_id(test_contact_id)
|
||||
name = f"{contact.get('firstname', '')} {contact.get('lastname', '')}"
|
||||
print_test("Get Contact by ID", True, f"Retrieved: {name.strip()}")
|
||||
|
||||
# Update contact
|
||||
updated = await client.update_contact(
|
||||
test_contact_id,
|
||||
email="updatedcontact@example.com"
|
||||
)
|
||||
print_test("Update Contact", True, "Email updated")
|
||||
|
||||
# Delete contact
|
||||
deleted = await client.delete_contact(test_contact_id)
|
||||
print_test("Delete Contact", True, f"Deleted ID: {test_contact_id}")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print_test("Contact Operations", False, str(e))
|
||||
# Try to clean up
|
||||
if test_contact_id:
|
||||
try:
|
||||
await client.delete_contact(test_contact_id)
|
||||
except:
|
||||
pass
|
||||
return False
|
||||
|
||||
|
||||
async def test_raw_api(client):
|
||||
"""Test raw API access."""
|
||||
print("\n5. Testing Raw API Access...")
|
||||
print_header("🔧 RAW API ACCESS")
|
||||
|
||||
try:
|
||||
# Make a raw API call
|
||||
print(" Making raw API call to /users endpoint...")
|
||||
# Test raw GET request
|
||||
result = await client.dolibarr_raw_api(
|
||||
method="GET",
|
||||
endpoint="/users",
|
||||
params={"limit": 2}
|
||||
endpoint="/setup/modules",
|
||||
params={"limit": 5}
|
||||
)
|
||||
print(f" ✅ Raw API call successful")
|
||||
print(f" Response type: {type(result)}")
|
||||
print_test("Raw GET Request", True, f"Retrieved {len(result) if isinstance(result, list) else 'data'}")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f" ❌ Raw API call failed: {e}")
|
||||
print_test("Raw API Access", False, str(e))
|
||||
return False
|
||||
|
||||
|
||||
async def main():
|
||||
"""Run all tests."""
|
||||
print("=" * 60)
|
||||
print("🧪 DOLIBARR MCP SERVER TEST SUITE")
|
||||
print("=" * 60)
|
||||
print(f"{TestColors.BOLD}{TestColors.HEADER}")
|
||||
print("╔══════════════════════════════════════════════════════════╗")
|
||||
print("║ 🚀 DOLIBARR MCP SERVER - COMPREHENSIVE TEST SUITE ║")
|
||||
print("╚══════════════════════════════════════════════════════════╝")
|
||||
print(f"{TestColors.ENDC}")
|
||||
|
||||
# Initialize configuration and client
|
||||
try:
|
||||
# Load configuration
|
||||
print("\n📋 Loading configuration...")
|
||||
config = Config()
|
||||
print(f" API URL: {config.dolibarr_url}")
|
||||
print(f" API Key: {'*' * 10 if config.api_key else 'NOT SET'}")
|
||||
|
||||
if not config.api_key or config.api_key == "your_api_key_here":
|
||||
print("\n❌ ERROR: Please configure your Dolibarr API credentials in .env file")
|
||||
print(" Required variables:")
|
||||
print(" - DOLIBARR_URL: Your Dolibarr instance URL")
|
||||
print(" - DOLIBARR_API_KEY: Your API key")
|
||||
return
|
||||
|
||||
# Create client and run tests
|
||||
async with DolibarrClient(config) as client:
|
||||
tests_passed = 0
|
||||
tests_total = 5
|
||||
|
||||
# Run test suite
|
||||
if await test_connection(client):
|
||||
tests_passed += 1
|
||||
|
||||
if await test_customer_crud(client):
|
||||
tests_passed += 1
|
||||
|
||||
if await test_product_crud(client):
|
||||
tests_passed += 1
|
||||
|
||||
if await test_list_operations(client):
|
||||
tests_passed += 1
|
||||
|
||||
if await test_raw_api(client):
|
||||
tests_passed += 1
|
||||
|
||||
# Summary
|
||||
print("\n" + "=" * 60)
|
||||
print("📊 TEST SUMMARY")
|
||||
print("=" * 60)
|
||||
print(f"Tests Passed: {tests_passed}/{tests_total}")
|
||||
|
||||
if tests_passed == tests_total:
|
||||
print("✅ ALL TESTS PASSED! The Dolibarr MCP Server is fully operational.")
|
||||
elif tests_passed > 0:
|
||||
print(f"⚠️ PARTIAL SUCCESS: {tests_passed} tests passed, {tests_total - tests_passed} failed.")
|
||||
else:
|
||||
print("❌ ALL TESTS FAILED. Please check your configuration and API access.")
|
||||
|
||||
print("\n📌 Next Steps:")
|
||||
print("1. If all tests passed, your MCP server is ready for production use")
|
||||
print("2. Run the server: python -m dolibarr_mcp.dolibarr_mcp_server")
|
||||
print("3. Or use Docker: docker-compose up")
|
||||
print("4. Connect your AI assistant to the MCP server")
|
||||
|
||||
print(f"\n{TestColors.OKGREEN}✅ Configuration loaded successfully{TestColors.ENDC}")
|
||||
print(f" URL: {config.dolibarr_url}")
|
||||
print(f" API Key: {'*' * 20}{config.api_key[-4:]}")
|
||||
except Exception as e:
|
||||
print(f"\n❌ FATAL ERROR: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
print(f"\n{TestColors.FAIL}❌ Configuration failed: {e}{TestColors.ENDC}")
|
||||
return
|
||||
|
||||
# Run tests
|
||||
async with DolibarrClient(config) as client:
|
||||
results = []
|
||||
|
||||
# Run each test suite
|
||||
results.append(("Connection", await test_connection(client)))
|
||||
results.append(("Users", await test_users(client)))
|
||||
results.append(("Customers", await test_customers(client)))
|
||||
results.append(("Products", await test_products(client)))
|
||||
results.append(("Invoices", await test_invoices(client)))
|
||||
results.append(("Orders", await test_orders(client)))
|
||||
results.append(("Contacts", await test_contacts(client)))
|
||||
results.append(("Raw API", await test_raw_api(client)))
|
||||
|
||||
# Print summary
|
||||
print_header("📊 TEST SUMMARY")
|
||||
|
||||
total_tests = len(results)
|
||||
passed_tests = sum(1 for _, passed in results if passed)
|
||||
failed_tests = total_tests - passed_tests
|
||||
|
||||
print(f"\n Total Tests: {total_tests}")
|
||||
print(f" {TestColors.OKGREEN}Passed: {passed_tests}{TestColors.ENDC}")
|
||||
print(f" {TestColors.FAIL}Failed: {failed_tests}{TestColors.ENDC}")
|
||||
|
||||
if failed_tests == 0:
|
||||
print(f"\n{TestColors.OKGREEN}{TestColors.BOLD}🎉 ALL TESTS PASSED! 🎉{TestColors.ENDC}")
|
||||
print(f"\n{TestColors.OKCYAN}The Dolibarr MCP server is fully operational!{TestColors.ENDC}")
|
||||
print(f"{TestColors.OKCYAN}You can now use it with Claude Desktop.{TestColors.ENDC}")
|
||||
else:
|
||||
print(f"\n{TestColors.WARNING}⚠️ Some tests failed. Check the details above.{TestColors.ENDC}")
|
||||
print(f"{TestColors.WARNING}The server may still work for successful operations.{TestColors.ENDC}")
|
||||
|
||||
print(f"\n{TestColors.BOLD}Next steps:{TestColors.ENDC}")
|
||||
print("1. Ensure Claude Desktop configuration includes this server")
|
||||
print("2. Restart Claude Desktop to reload MCP servers")
|
||||
print("3. Test by asking Claude to interact with your Dolibarr system")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
try:
|
||||
asyncio.run(main())
|
||||
except KeyboardInterrupt:
|
||||
print(f"\n\n{TestColors.WARNING}Test suite interrupted by user{TestColors.ENDC}")
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(f"\n{TestColors.FAIL}Unexpected error: {e}{TestColors.ENDC}")
|
||||
sys.exit(1)
|
||||
|
||||
Reference in New Issue
Block a user