""" CRUD Operations Integration Tests for Dolibarr MCP Server. These tests verify complete CRUD operations for all Dolibarr entities. Run with: pytest tests/test_crud_operations.py -v """ import pytest import asyncio from datetime import datetime from unittest.mock import Mock, patch, AsyncMock from dolibarr_mcp import DolibarrClient, DolibarrMCPServer class TestCRUDOperations: """Test complete CRUD operations for all Dolibarr entities.""" @pytest.fixture def client(self): """Create a test client instance.""" return DolibarrClient( url="https://test.dolibarr.com", api_key="test_api_key" ) @pytest.fixture def server(self): """Create a test server instance.""" with patch('dolibarr_mcp.dolibarr_mcp_server.Config') as mock_config: mock_config.return_value.dolibarr_url = "https://test.dolibarr.com" mock_config.return_value.dolibarr_api_key = "test_api_key" mock_config.return_value.log_level = "INFO" return DolibarrMCPServer() # Customer (Third Party) CRUD Tests @pytest.mark.asyncio async def test_customer_crud_lifecycle(self, client): """Test complete customer CRUD lifecycle.""" # Mock responses for each operation with patch.object(client, 'request') as mock_request: # Create mock_request.return_value = {"id": 1} customer_id = await client.create_customer({ "name": "Test Company", "email": "test@company.com" }) assert customer_id == 1 # Read mock_request.return_value = { "id": 1, "name": "Test Company", "email": "test@company.com" } customer = await client.get_customer_by_id(1) assert customer["name"] == "Test Company" # Update mock_request.return_value = {"id": 1, "name": "Updated Company"} result = await client.update_customer(1, {"name": "Updated Company"}) assert result["name"] == "Updated Company" # Delete mock_request.return_value = {"success": True} result = await client.delete_customer(1) assert result["success"] is True # Product CRUD Tests @pytest.mark.asyncio async def test_product_crud_lifecycle(self, client): """Test complete product CRUD lifecycle.""" with patch.object(client, 'request') as mock_request: # Create mock_request.return_value = {"id": 10} product_id = await client.create_product({ "label": "Test Product", "price": 99.99, "description": "Test product description" }) assert product_id == 10 # Read mock_request.return_value = { "id": 10, "label": "Test Product", "price": "99.99" } product = await client.get_product_by_id(10) assert product["label"] == "Test Product" # Update mock_request.return_value = {"id": 10, "price": "149.99"} result = await client.update_product(10, {"price": 149.99}) assert result["price"] == "149.99" # Delete mock_request.return_value = {"success": True} result = await client.delete_product(10) assert result["success"] is True # Invoice CRUD Tests @pytest.mark.asyncio async def test_invoice_crud_lifecycle(self, client): """Test complete invoice CRUD lifecycle.""" with patch.object(client, 'request') as mock_request: # Create mock_request.return_value = {"id": 100} invoice_id = await client.create_invoice({ "socid": 1, # Customer ID "date": datetime.now().isoformat(), "lines": [ {"desc": "Service", "qty": 1, "subprice": 100} ] }) assert invoice_id == 100 # Read mock_request.return_value = { "id": 100, "ref": "INV-2024-100", "total_ttc": "100.00" } invoice = await client.get_invoice_by_id(100) assert invoice["ref"] == "INV-2024-100" # Update mock_request.return_value = {"id": 100, "date_lim_reglement": "2024-02-01"} result = await client.update_invoice(100, { "date_lim_reglement": "2024-02-01" }) assert result["date_lim_reglement"] == "2024-02-01" # Delete mock_request.return_value = {"success": True} result = await client.delete_invoice(100) assert result["success"] is True # Order CRUD Tests @pytest.mark.asyncio async def test_order_crud_lifecycle(self, client): """Test complete order CRUD lifecycle.""" with patch.object(client, 'request') as mock_request: # Create mock_request.return_value = {"id": 50} order_id = await client.create_order({ "socid": 1, "date": datetime.now().isoformat() }) assert order_id == 50 # Read mock_request.return_value = { "id": 50, "ref": "ORD-2024-50", "socid": 1 } order = await client.get_order_by_id(50) assert order["ref"] == "ORD-2024-50" # Update mock_request.return_value = {"id": 50, "note_public": "Updated note"} result = await client.update_order(50, { "note_public": "Updated note" }) assert result["note_public"] == "Updated note" # Delete mock_request.return_value = {"success": True} result = await client.delete_order(50) assert result["success"] is True # Contact CRUD Tests @pytest.mark.asyncio async def test_contact_crud_lifecycle(self, client): """Test complete contact CRUD lifecycle.""" with patch.object(client, 'request') as mock_request: # Create mock_request.return_value = {"id": 20} contact_id = await client.create_contact({ "firstname": "John", "lastname": "Doe", "email": "john.doe@example.com" }) assert contact_id == 20 # Read mock_request.return_value = { "id": 20, "firstname": "John", "lastname": "Doe", "email": "john.doe@example.com" } contact = await client.get_contact_by_id(20) assert contact["firstname"] == "John" # Update mock_request.return_value = {"id": 20, "phone": "+1234567890"} result = await client.update_contact(20, { "phone": "+1234567890" }) assert result["phone"] == "+1234567890" # Delete mock_request.return_value = {"success": True} result = await client.delete_contact(20) assert result["success"] is True # User Management Tests @pytest.mark.asyncio async def test_user_crud_lifecycle(self, client): """Test complete user CRUD lifecycle.""" with patch.object(client, 'request') as mock_request: # Create mock_request.return_value = {"id": 5} user_id = await client.create_user({ "login": "testuser", "lastname": "User", "firstname": "Test", "email": "testuser@example.com" }) assert user_id == 5 # Read mock_request.return_value = { "id": 5, "login": "testuser", "email": "testuser@example.com" } user = await client.get_user_by_id(5) assert user["login"] == "testuser" # Update mock_request.return_value = {"id": 5, "admin": 1} result = await client.update_user(5, {"admin": 1}) assert result["admin"] == 1 # Delete mock_request.return_value = {"success": True} result = await client.delete_user(5) assert result["success"] is True # Batch Operations Tests @pytest.mark.asyncio async def test_batch_operations(self, client): """Test batch operations for multiple entities.""" with patch.object(client, 'request') as mock_request: # Get multiple customers mock_request.return_value = [ {"id": 1, "name": "Company A"}, {"id": 2, "name": "Company B"}, {"id": 3, "name": "Company C"} ] customers = await client.get_customers(limit=3) assert len(customers) == 3 # Get multiple products mock_request.return_value = [ {"id": 10, "label": "Product A"}, {"id": 11, "label": "Product B"} ] products = await client.get_products(limit=2) assert len(products) == 2 # Error Handling Tests @pytest.mark.asyncio async def test_error_handling(self, client): """Test error handling in CRUD operations.""" with patch.object(client, 'request') as mock_request: # Test 404 Not Found mock_request.side_effect = Exception("404 Not Found") with pytest.raises(Exception, match="404"): await client.get_customer_by_id(999) # Test validation error mock_request.side_effect = Exception("Validation Error: Missing required field") with pytest.raises(Exception, match="Validation"): await client.create_product({}) # Test connection error mock_request.side_effect = Exception("Connection refused") with pytest.raises(Exception, match="Connection"): await client.test_connection() class TestMCPServerIntegration: """Test MCP Server integration with Dolibarr API.""" @pytest.mark.asyncio async def test_server_initialization(self): """Test server initialization and configuration.""" with patch('dolibarr_mcp.config.Config') as mock_config: mock_config.return_value.dolibarr_url = "https://test.com" mock_config.return_value.dolibarr_api_key = "key" mock_config.return_value.log_level = "INFO" server = DolibarrMCPServer() assert server is not None assert hasattr(server, 'client') @pytest.mark.asyncio async def test_server_tool_execution(self): """Test server tool execution.""" with patch('dolibarr_mcp.dolibarr_mcp_server.DolibarrClient') as mock_client: server = DolibarrMCPServer() # Mock tool execution mock_client.return_value.get_customers = AsyncMock(return_value=[ {"id": 1, "name": "Test Company"} ]) # Simulate tool call result = await server.client.get_customers() assert len(result) == 1 assert result[0]["name"] == "Test Company" if __name__ == "__main__": # Run tests with coverage pytest.main([__file__, "-v", "--cov=dolibarr_mcp", "--cov-report=term-missing"])