"""
Тесты для n8n MCP-сервера.
Tests for n8n MCP server.
"""
import pytest
from unittest.mock import Mock, patch
from n8n_mcp_server.n8n_client import N8NClient
from n8n_mcp_server.tools import (
create_workflow_tool,
delete_workflow_tool,
execute_workflow_tool,
get_execution_status_tool,
get_workflow_tool,
list_workflows_tool,
update_workflow_tool,
)
class TestN8NClient:
"""Тесты для N8NClient."""
@patch('n8n_mcp_server.n8n_client.requests.Session')
def test_list_workflows(self, mock_session):
"""Тест получения списка workflow."""
# Мокаем ответ API
mock_response = Mock()
mock_response.json.return_value = [
{"id": "1", "name": "Test Workflow", "active": True}
]
mock_session.return_value.request.return_value = mock_response
client = N8NClient()
result = client.list_workflows()
assert isinstance(result, list)
assert len(result) == 1
assert result[0]["name"] == "Test Workflow"
@patch('n8n_mcp_server.n8n_client.requests.Session')
def test_get_workflow(self, mock_session):
"""Тест получения workflow по ID."""
mock_response = Mock()
mock_response.json.return_value = {
"id": "1",
"name": "Test Workflow",
"active": True
}
mock_session.return_value.request.return_value = mock_response
client = N8NClient()
result = client.get_workflow("1")
assert result["name"] == "Test Workflow"
mock_session.return_value.request.assert_called_once()
@patch('n8n_mcp_server.n8n_client.requests.Session')
def test_create_workflow(self, mock_session):
"""Тест создания workflow."""
mock_response = Mock()
mock_response.json.return_value = {
"id": "1",
"name": "New Workflow"
}
mock_session.return_value.request.return_value = mock_response
client = N8NClient()
workflow_data = {
"name": "New Workflow",
"active": False
}
result = client.create_workflow(workflow_data)
assert result["name"] == "New Workflow"
# Проверяем, что был сделан POST-запрос
call_args = mock_session.return_value.request.call_args
assert call_args[1]["method"] == "POST"
class TestTools:
"""Тесты для инструментов."""
def test_list_workflows_tool(self):
"""Тест инструмента list_workflows."""
client = Mock(spec=N8NClient)
client.list_workflows.return_value = [
{"id": "1", "name": "Test Workflow"}
]
tool = list_workflows_tool(client)
result = tool.execute()
assert len(result) == 1
assert result[0]["name"] == "Test Workflow"
def test_get_workflow_tool(self):
"""Тест инструмента get_workflow."""
client = Mock(spec=N8NClient)
client.get_workflow.return_value = {
"id": "1",
"name": "Test Workflow"
}
tool = get_workflow_tool(client)
result = tool.execute(workflow_id="1")
assert result["name"] == "Test Workflow"
def test_create_workflow_tool(self):
"""Тест инструмента create_workflow."""
client = Mock(spec=N8NClient)
client.create_workflow.return_value = {
"id": "1",
"name": "New Workflow"
}
tool = create_workflow_tool(client)
result = tool.execute(
name="New Workflow",
active=False
)
assert result["name"] == "New Workflow"
def test_execute_workflow_tool(self):
"""Тест инструмента execute_workflow."""
client = Mock(spec=N8NClient)
client.execute_workflow.return_value = {
"executionId": "exec_123",
"status": "running"
}
tool = execute_workflow_tool(client)
result = tool.execute(
workflow_id="1",
input_data={"test": "data"}
)
assert result["executionId"] == "exec_123"
def test_get_execution_status_tool(self):
"""Тест инструмента get_execution_status."""
client = Mock(spec=N8NClient)
client.get_execution_status.return_value = {
"id": "exec_123",
"status": "completed"
}
tool = get_execution_status_tool(client)
result = tool.execute(execution_id="exec_123")
assert result["status"] == "completed"
class TestErrorHandling:
"""Тесты обработки ошибок."""
def test_client_request_exception(self):
"""Тест обработки исключений при запросах."""
with patch('n8n_mcp_server.n8n_client.requests.Session') as mock_session:
mock_session.return_value.request.side_effect = Exception("API Error")
client = N8NClient()
with pytest.raises(Exception) as exc_info:
client.list_workflows()
assert "API Error" in str(exc_info.value)
def test_tool_with_client_error(self):
"""Тест обработки ошибок в инструментах."""
client = Mock(spec=N8NClient)
client.list_workflows.side_effect = Exception("Client error")
tool = list_workflows_tool(client)
with pytest.raises(Exception) as exc_info:
tool.execute()
assert "Client error" in str(exc_info.value)
