diff --git a/tests/test_crud_operations.py b/tests/test_crud_operations.py new file mode 100644 index 0000000..42a1c04 --- /dev/null +++ b/tests/test_crud_operations.py @@ -0,0 +1,328 @@ +""" +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"])