From 4d72cc735513501783fe28ddb88a18f2daa35354 Mon Sep 17 00:00:00 2001
From: latinogino <154024559+latinogino@users.noreply.github.com>
Date: Tue, 23 Sep 2025 16:08:26 +0200
Subject: [PATCH] Add comprehensive test suite for all Dolibarr MCP operations

---
 test_dolibarr_mcp.py | 456 +++++++++++++++++++++++++++----------------
 1 file changed, 292 insertions(+), 164 deletions(-)

diff --git a/test_dolibarr_mcp.py b/test_dolibarr_mcp.py
index 058ae61..53b17c6 100644
--- a/test_dolibarr_mcp.py
+++ b/test_dolibarr_mcp.py
@@ -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)