ServiceNow MCP Server

""" Tests for the workflow management tools. """ import json import unittest from datetime import datetime, timedelta from unittest.mock import MagicMock, patch import requests from servicenow_mcp.auth.auth_manager import AuthManager from servicenow_mcp.tools.workflow_tools import ( list_workflows, get_workflow_details, list_workflow_versions, get_workflow_activities, create_workflow, update_workflow, activate_workflow, deactivate_workflow, add_workflow_activity, update_workflow_activity, delete_workflow_activity, reorder_workflow_activities, ) from servicenow_mcp.utils.config import AuthConfig, AuthType, BasicAuthConfig, ServerConfig class TestWorkflowTools(unittest.TestCase): """Tests for the workflow management tools.""" def setUp(self): """Set up test fixtures.""" self.auth_config = AuthConfig( type=AuthType.BASIC, basic=BasicAuthConfig(username="test_user", password="test_password"), ) self.server_config = ServerConfig( instance_url="https://test.service-now.com", auth=self.auth_config, ) self.auth_manager = AuthManager(self.auth_config) @patch("servicenow_mcp.tools.workflow_tools.requests.get") def test_list_workflows_success(self, mock_get): """Test listing workflows successfully.""" # Mock the response mock_response = MagicMock() mock_response.json.return_value = { "result": [ { "sys_id": "workflow123", "name": "Incident Approval", "description": "Workflow for incident approval", "active": "true", "table": "incident", }, { "sys_id": "workflow456", "name": "Change Request", "description": "Workflow for change requests", "active": "true", "table": "change_request", }, ] } mock_response.headers = {"X-Total-Count": "2"} mock_response.raise_for_status = MagicMock() mock_get.return_value = mock_response # Call the function params = { "limit": 10, "active": True, } result = list_workflows(self.auth_manager, self.server_config, params) # Verify the result self.assertEqual(len(result["workflows"]), 2) self.assertEqual(result["count"], 2) self.assertEqual(result["total"], 2) self.assertEqual(result["workflows"][0]["sys_id"], "workflow123") self.assertEqual(result["workflows"][1]["sys_id"], "workflow456") @patch("servicenow_mcp.tools.workflow_tools.requests.get") def test_list_workflows_empty_result(self, mock_get): """Test listing workflows with empty result.""" # Mock the response mock_response = MagicMock() mock_response.json.return_value = {"result": []} mock_response.headers = {"X-Total-Count": "0"} mock_response.raise_for_status = MagicMock() mock_get.return_value = mock_response # Call the function params = { "limit": 10, "active": True, } result = list_workflows(self.auth_manager, self.server_config, params) # Verify the result self.assertEqual(len(result["workflows"]), 0) self.assertEqual(result["count"], 0) self.assertEqual(result["total"], 0) @patch("servicenow_mcp.tools.workflow_tools.requests.get") def test_list_workflows_error(self, mock_get): """Test listing workflows with error.""" # Mock the response mock_get.side_effect = requests.RequestException("API Error") # Call the function params = { "limit": 10, "active": True, } result = list_workflows(self.auth_manager, self.server_config, params) # Verify the result self.assertIn("error", result) self.assertEqual(result["error"], "API Error") @patch("servicenow_mcp.tools.workflow_tools.requests.get") def test_get_workflow_details_success(self, mock_get): """Test getting workflow details successfully.""" # Mock the response mock_response = MagicMock() mock_response.json.return_value = { "result": { "sys_id": "workflow123", "name": "Incident Approval", "description": "Workflow for incident approval", "active": "true", "table": "incident", } } mock_response.raise_for_status = MagicMock() mock_get.return_value = mock_response # Call the function params = { "workflow_id": "workflow123", } result = get_workflow_details(self.auth_manager, self.server_config, params) # Verify the result self.assertEqual(result["workflow"]["sys_id"], "workflow123") self.assertEqual(result["workflow"]["name"], "Incident Approval") @patch("servicenow_mcp.tools.workflow_tools.requests.get") def test_get_workflow_details_error(self, mock_get): """Test getting workflow details with error.""" # Mock the response mock_get.side_effect = requests.RequestException("API Error") # Call the function params = { "workflow_id": "workflow123", } result = get_workflow_details(self.auth_manager, self.server_config, params) # Verify the result self.assertIn("error", result) self.assertEqual(result["error"], "API Error") @patch("servicenow_mcp.tools.workflow_tools.requests.get") def test_list_workflow_versions_success(self, mock_get): """Test listing workflow versions successfully.""" # Mock the response mock_response = MagicMock() mock_response.json.return_value = { "result": [ { "sys_id": "version123", "workflow": "workflow123", "name": "Version 1", "version": "1", "published": "true", }, { "sys_id": "version456", "workflow": "workflow123", "name": "Version 2", "version": "2", "published": "true", }, ] } mock_response.headers = {"X-Total-Count": "2"} mock_response.raise_for_status = MagicMock() mock_get.return_value = mock_response # Call the function params = { "workflow_id": "workflow123", "limit": 10, } result = list_workflow_versions(self.auth_manager, self.server_config, params) # Verify the result self.assertEqual(len(result["versions"]), 2) self.assertEqual(result["count"], 2) self.assertEqual(result["total"], 2) self.assertEqual(result["versions"][0]["sys_id"], "version123") self.assertEqual(result["versions"][1]["sys_id"], "version456") @patch("servicenow_mcp.tools.workflow_tools.requests.get") def test_get_workflow_activities_success(self, mock_get): """Test getting workflow activities successfully.""" # Mock the responses for version query and activities query version_response = MagicMock() version_response.json.return_value = { "result": [ { "sys_id": "version123", "workflow": "workflow123", "name": "Version 1", "version": "1", "published": "true", } ] } version_response.raise_for_status = MagicMock() activities_response = MagicMock() activities_response.json.return_value = { "result": [ { "sys_id": "activity123", "workflow_version": "version123", "name": "Approval", "order": "100", "activity_definition": "approval", }, { "sys_id": "activity456", "workflow_version": "version123", "name": "Notification", "order": "200", "activity_definition": "notification", }, ] } activities_response.raise_for_status = MagicMock() # Configure the mock to return different responses for different URLs def side_effect(*args, **kwargs): url = args[0] if args else kwargs.get('url', '') if 'wf_workflow_version' in url: return version_response elif 'wf_activity' in url: return activities_response return MagicMock() mock_get.side_effect = side_effect # Call the function params = { "workflow_id": "workflow123", } result = get_workflow_activities(self.auth_manager, self.server_config, params) # Verify the result self.assertEqual(len(result["activities"]), 2) self.assertEqual(result["count"], 2) self.assertEqual(result["workflow_id"], "workflow123") self.assertEqual(result["version_id"], "version123") self.assertEqual(result["activities"][0]["sys_id"], "activity123") self.assertEqual(result["activities"][1]["sys_id"], "activity456") @patch("servicenow_mcp.tools.workflow_tools.requests.post") def test_create_workflow_success(self, mock_post): """Test creating a workflow successfully.""" # Mock the response mock_response = MagicMock() mock_response.json.return_value = { "result": { "sys_id": "workflow789", "name": "New Workflow", "description": "A new workflow", "active": "true", "table": "incident", } } mock_response.raise_for_status = MagicMock() mock_post.return_value = mock_response # Call the function params = { "name": "New Workflow", "description": "A new workflow", "table": "incident", "active": True, } result = create_workflow(self.auth_manager, self.server_config, params) # Verify the result self.assertEqual(result["workflow"]["sys_id"], "workflow789") self.assertEqual(result["workflow"]["name"], "New Workflow") self.assertEqual(result["message"], "Workflow created successfully") @patch("servicenow_mcp.tools.workflow_tools.requests.patch") def test_update_workflow_success(self, mock_patch): """Test updating a workflow successfully.""" # Mock the response mock_response = MagicMock() mock_response.json.return_value = { "result": { "sys_id": "workflow123", "name": "Updated Workflow", "description": "Updated description", "active": "true", "table": "incident", } } mock_response.raise_for_status = MagicMock() mock_patch.return_value = mock_response # Call the function params = { "workflow_id": "workflow123", "name": "Updated Workflow", "description": "Updated description", } result = update_workflow(self.auth_manager, self.server_config, params) # Verify the result self.assertEqual(result["workflow"]["sys_id"], "workflow123") self.assertEqual(result["workflow"]["name"], "Updated Workflow") self.assertEqual(result["message"], "Workflow updated successfully") @patch("servicenow_mcp.tools.workflow_tools.requests.patch") def test_activate_workflow_success(self, mock_patch): """Test activating a workflow successfully.""" # Mock the response mock_response = MagicMock() mock_response.json.return_value = { "result": { "sys_id": "workflow123", "name": "Incident Approval", "active": "true", } } mock_response.raise_for_status = MagicMock() mock_patch.return_value = mock_response # Call the function params = { "workflow_id": "workflow123", } result = activate_workflow(self.auth_manager, self.server_config, params) # Verify the result self.assertEqual(result["workflow"]["sys_id"], "workflow123") self.assertEqual(result["workflow"]["active"], "true") self.assertEqual(result["message"], "Workflow activated successfully") @patch("servicenow_mcp.tools.workflow_tools.requests.patch") def test_deactivate_workflow_success(self, mock_patch): """Test deactivating a workflow successfully.""" # Mock the response mock_response = MagicMock() mock_response.json.return_value = { "result": { "sys_id": "workflow123", "name": "Incident Approval", "active": "false", } } mock_response.raise_for_status = MagicMock() mock_patch.return_value = mock_response # Call the function params = { "workflow_id": "workflow123", } result = deactivate_workflow(self.auth_manager, self.server_config, params) # Verify the result self.assertEqual(result["workflow"]["sys_id"], "workflow123") self.assertEqual(result["workflow"]["active"], "false") self.assertEqual(result["message"], "Workflow deactivated successfully") @patch("servicenow_mcp.tools.workflow_tools.requests.get") @patch("servicenow_mcp.tools.workflow_tools.requests.post") def test_add_workflow_activity_success(self, mock_post, mock_get): """Test adding a workflow activity successfully.""" # Mock the responses for version query and activity creation version_response = MagicMock() version_response.json.return_value = { "result": [ { "sys_id": "version123", "workflow": "workflow123", "name": "Version 1", "version": "1", "published": "false", } ] } version_response.raise_for_status = MagicMock() order_response = MagicMock() order_response.json.return_value = { "result": [ { "sys_id": "activity123", "order": "100", } ] } order_response.raise_for_status = MagicMock() activity_response = MagicMock() activity_response.json.return_value = { "result": { "sys_id": "activity789", "workflow_version": "version123", "name": "New Activity", "order": "200", "activity_definition": "approval", } } activity_response.raise_for_status = MagicMock() # Configure the mocks def get_side_effect(*args, **kwargs): url = args[0] if args else kwargs.get('url', '') if 'wf_workflow_version' in url: return version_response elif 'wf_activity' in url: return order_response return MagicMock() mock_get.side_effect = get_side_effect mock_post.return_value = activity_response # Call the function params = { "workflow_id": "workflow123", "name": "New Activity", "activity_type": "approval", "description": "A new approval activity", } result = add_workflow_activity(self.auth_manager, self.server_config, params) # Verify the result self.assertEqual(result["activity"]["sys_id"], "activity789") self.assertEqual(result["activity"]["name"], "New Activity") self.assertEqual(result["workflow_id"], "workflow123") self.assertEqual(result["version_id"], "version123") self.assertEqual(result["message"], "Activity added successfully") @patch("servicenow_mcp.tools.workflow_tools.requests.patch") def test_update_workflow_activity_success(self, mock_patch): """Test updating a workflow activity successfully.""" # Mock the response mock_response = MagicMock() mock_response.json.return_value = { "result": { "sys_id": "activity123", "name": "Updated Activity", "description": "Updated description", } } mock_response.raise_for_status = MagicMock() mock_patch.return_value = mock_response # Call the function params = { "activity_id": "activity123", "name": "Updated Activity", "description": "Updated description", } result = update_workflow_activity(self.auth_manager, self.server_config, params) # Verify the result self.assertEqual(result["activity"]["sys_id"], "activity123") self.assertEqual(result["activity"]["name"], "Updated Activity") self.assertEqual(result["message"], "Activity updated successfully") @patch("servicenow_mcp.tools.workflow_tools.requests.delete") def test_delete_workflow_activity_success(self, mock_delete): """Test deleting a workflow activity successfully.""" # Mock the response mock_response = MagicMock() mock_response.raise_for_status = MagicMock() mock_delete.return_value = mock_response # Call the function params = { "activity_id": "activity123", } result = delete_workflow_activity(self.auth_manager, self.server_config, params) # Verify the result self.assertEqual(result["message"], "Activity deleted successfully") self.assertEqual(result["activity_id"], "activity123") @patch("servicenow_mcp.tools.workflow_tools.requests.patch") def test_reorder_workflow_activities_success(self, mock_patch): """Test reordering workflow activities successfully.""" # Mock the response mock_response = MagicMock() mock_response.json.return_value = {"result": {}} mock_response.raise_for_status = MagicMock() mock_patch.return_value = mock_response # Call the function params = { "workflow_id": "workflow123", "activity_ids": ["activity1", "activity2", "activity3"], } result = reorder_workflow_activities(self.auth_manager, self.server_config, params) # Verify the result self.assertEqual(result["message"], "Activities reordered") self.assertEqual(result["workflow_id"], "workflow123") self.assertEqual(len(result["results"]), 3) self.assertTrue(all(item["success"] for item in result["results"])) self.assertEqual(result["results"][0]["activity_id"], "activity1") self.assertEqual(result["results"][0]["new_order"], 100) self.assertEqual(result["results"][1]["activity_id"], "activity2") self.assertEqual(result["results"][1]["new_order"], 200) self.assertEqual(result["results"][2]["activity_id"], "activity3") self.assertEqual(result["results"][2]["new_order"], 300) if __name__ == "__main__": unittest.main()