feat: add pytest testing infrastructure
- Add test optional dependencies to pyproject.toml - Add conftest.py with fixtures for mocking settings and MCP client - Add test_server.py with tests for tryton_call and MCP tools
This commit is contained in:
25
tests/conftest.py
Normal file
25
tests/conftest.py
Normal file
@@ -0,0 +1,25 @@
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from unittest.mock import MagicMock, patch
|
||||
from fastmcp import Client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_settings():
|
||||
with patch("tryton_mcp.server.settings") as mock:
|
||||
mock_client = MagicMock()
|
||||
mock.get_client.return_value = mock_client
|
||||
yield mock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mcp_server():
|
||||
from tryton_mcp.server import mcp
|
||||
|
||||
return mcp
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def mcp_client(mcp_server):
|
||||
async with Client(mcp_server) as client:
|
||||
yield client
|
||||
141
tests/test_server.py
Normal file
141
tests/test_server.py
Normal file
@@ -0,0 +1,141 @@
|
||||
import pytest
|
||||
from unittest.mock import MagicMock, patch
|
||||
import datetime
|
||||
import json
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent / "src"))
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestTrytonCall:
|
||||
async def test_tryton_call_success(self, mock_settings):
|
||||
from tryton_mcp.server import tryton_call
|
||||
|
||||
mock_client = mock_settings.get_client.return_value
|
||||
mock_client.call.return_value = [{"id": 1, "name": "Test"}]
|
||||
|
||||
result = tryton_call("model.test.read", [[1], ["name"]])
|
||||
|
||||
assert result["success"] is True
|
||||
assert result["data"] == [{"id": 1, "name": "Test"}]
|
||||
mock_client.call.assert_called_once_with("model.test.read", [[1], ["name"]])
|
||||
|
||||
async def test_tryton_call_failure(self, mock_settings):
|
||||
from tryton_mcp.server import tryton_call
|
||||
|
||||
mock_client = mock_settings.get_client.return_value
|
||||
mock_client.call.side_effect = Exception("Connection error")
|
||||
|
||||
result = tryton_call("model.test.read", [[1], ["name"]])
|
||||
|
||||
assert result["success"] is False
|
||||
assert "Connection error" in result["error"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestCreateSchedule:
|
||||
async def test_create_schedule_success(self, mcp_client, mock_settings):
|
||||
from tryton_mcp.server import create_schedule
|
||||
|
||||
mock_client = mock_settings.get_client.return_value
|
||||
mock_client.call.return_value = [1]
|
||||
|
||||
result = await mcp_client.call_tool(
|
||||
"create_schedule",
|
||||
{
|
||||
"professional": 6,
|
||||
"description": "Test appointment",
|
||||
"customer": 4,
|
||||
"date": "2026-03-15T10:00:00",
|
||||
"service_center": 11,
|
||||
},
|
||||
)
|
||||
|
||||
assert result.content is not None
|
||||
assert len(result.content) == 1
|
||||
assert result.content[0].text == "[1]"
|
||||
|
||||
mock_client.call.assert_called_once()
|
||||
|
||||
async def test_create_schedule_failure(self, mcp_client, mock_settings):
|
||||
from tryton_mcp.server import create_schedule
|
||||
|
||||
mock_client = mock_settings.get_client.return_value
|
||||
mock_client.call.side_effect = Exception("Create failed")
|
||||
|
||||
with pytest.raises(Exception, match="Failed to create schedule"):
|
||||
await mcp_client.call_tool(
|
||||
"create_schedule",
|
||||
{
|
||||
"professional": 6,
|
||||
"description": "Test appointment",
|
||||
"customer": 4,
|
||||
"date": "2026-03-15T10:00:00",
|
||||
"service_center": 11,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestFindServiceCenters:
|
||||
async def test_find_service_centers_success(self, mcp_client, mock_settings):
|
||||
from tryton_mcp.server import find_service_centers
|
||||
|
||||
mock_client = mock_settings.get_client.return_value
|
||||
mock_client.call.return_value = [
|
||||
{"id": 11, "name": "Center 1", "address.street": "Street 1"}
|
||||
]
|
||||
|
||||
result = await mcp_client.call_tool("find_service_centers", {})
|
||||
|
||||
result_data = json.loads(result.content[0].text)
|
||||
assert result_data == [
|
||||
{"id": 11, "name": "Center 1", "address.street": "Street 1"}
|
||||
]
|
||||
call_args = mock_client.call.call_args[0]
|
||||
assert call_args[0] == "model.naliia.service_center.search_read"
|
||||
assert call_args[1][0] == [[]]
|
||||
|
||||
async def test_find_service_centers_failure(self, mcp_client, mock_settings):
|
||||
from tryton_mcp.server import find_service_centers
|
||||
|
||||
mock_client = mock_settings.get_client.return_value
|
||||
mock_client.call.side_effect = Exception("Search failed")
|
||||
|
||||
with pytest.raises(Exception, match="Failed to find service centers"):
|
||||
await mcp_client.call_tool("find_service_centers", {})
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestFindProductsAndServices:
|
||||
async def test_find_products_and_services_success(self, mcp_client, mock_settings):
|
||||
from tryton_mcp.server import find_products_and_services
|
||||
|
||||
mock_client = mock_settings.get_client.return_value
|
||||
mock_client.call.return_value = [
|
||||
{"id": 1, "name": "Product 1", "list_price": 100.0}
|
||||
]
|
||||
|
||||
result = await mcp_client.call_tool("find_products_and_services", {})
|
||||
|
||||
result_data = json.loads(result.content[0].text)
|
||||
assert result_data == [{"id": 1, "name": "Product 1", "list_price": 100.0}]
|
||||
|
||||
mock_client.call.assert_called_once()
|
||||
|
||||
call_args = mock_client.call.call_args[0]
|
||||
|
||||
assert call_args[0] == "model.product.product.search_read"
|
||||
assert call_args[1][0] == [["active", "=", True], ["salable", "=", True]]
|
||||
|
||||
async def test_find_products_and_services_failure(self, mcp_client, mock_settings):
|
||||
from tryton_mcp.server import find_products_and_services
|
||||
|
||||
mock_client = mock_settings.get_client.return_value
|
||||
mock_client.call.side_effect = Exception("Search failed")
|
||||
|
||||
with pytest.raises(Exception, match="Failed to find products"):
|
||||
await mcp_client.call_tool("find_products_and_services", {})
|
||||
Reference in New Issue
Block a user