diff --git a/pyproject.toml b/pyproject.toml index 609eaaa..3309aaf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -8,6 +8,7 @@ dependencies = [ "fastmcp[tasks]>=3.1.0", "sabatron-tryton-rpc-client>=7.4.0", "python-dotenv>=1.0.0", + "pydantic>=2.0.0", ] [project.optional-dependencies] diff --git a/src/tryton_mcp/__init__.py b/src/tryton_mcp/__init__.py index e69de29..0deff75 100644 --- a/src/tryton_mcp/__init__.py +++ b/src/tryton_mcp/__init__.py @@ -0,0 +1,4 @@ +from tryton_mcp.config import settings, TrytonSettings +from tryton_mcp.server import mcp + +__all__ = ["settings", "TrytonSettings", "mcp"] diff --git a/src/tryton_mcp/config.py b/src/tryton_mcp/config.py index 8d4da9f..0880605 100644 --- a/src/tryton_mcp/config.py +++ b/src/tryton_mcp/config.py @@ -1,4 +1,5 @@ import os +import threading from contextlib import asynccontextmanager from dataclasses import dataclass, field from pathlib import Path @@ -13,6 +14,10 @@ load_dotenv(project_root / ".env") @dataclass class TrytonSettings: + _lock = threading.Lock() + _instance = None + _client: Optional[Client] = field(default=None, init=False, repr=False) + hostname: str = field( default_factory=lambda: os.environ.get("TRYTON_HOSTNAME", "localhost") ) @@ -26,7 +31,6 @@ class TrytonSettings: port: int = field( default_factory=lambda: int(os.environ.get("TRYTON_PORT", "8000")) ) - _client: Optional[Client] = field(default=None, init=False, repr=False) def to_dict(self) -> dict: return { @@ -39,7 +43,9 @@ class TrytonSettings: def get_client(self) -> Client: if self._client is None: - self._client = Client(**self.to_dict()) + with self._lock: + if self._client is None: + self._client = Client(**self.to_dict()) return self._client def connect(self): diff --git a/src/tryton_mcp/server.py b/src/tryton_mcp/server.py index d1bd0d3..d111fc6 100644 --- a/src/tryton_mcp/server.py +++ b/src/tryton_mcp/server.py @@ -2,13 +2,16 @@ MCP Server for Naliia Module """ +import logging from contextlib import asynccontextmanager -from fastmcp import FastMCP -from typing import List, Dict, AsyncIterator, Any from datetime import datetime, timezone +from typing import Any, Dict, List, Optional + +from fastmcp import FastMCP +from pydantic import BaseModel, Field, field_validator from tryton_mcp.config import settings -import logging +from tryton_mcp.services.provider import ServiceProvider logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" @@ -18,7 +21,7 @@ logger = logging.getLogger(__name__) @asynccontextmanager -async def server_lifespan(server: FastMCP) -> AsyncIterator[None]: +async def server_lifespan(server: FastMCP): settings.connect() yield settings.disconnect() @@ -26,112 +29,132 @@ async def server_lifespan(server: FastMCP) -> AsyncIterator[None]: mcp = FastMCP("Tryton MCP Server", lifespan=server_lifespan) +_service_provider: Optional[ServiceProvider] = None -def tryton_call(name: str, args: list) -> Dict[str, Any]: - try: - client = settings.get_client() - response = client.call(name, args) - return {"success": True, "data": response} - except Exception as e: - return {"success": False, "error": str(e)} + +def get_service_provider() -> ServiceProvider: + global _service_provider + if _service_provider is None: + _service_provider = ServiceProvider(settings.get_client()) + return _service_provider + + +class FindCustomerInput(BaseModel): + identifier: str = Field(min_length=1, description="Customer identifier code") + + +class CreateCustomerInput(BaseModel): + name: str = Field(min_length=1, description="Customer name") + identifiers: List[tuple[str, List[Dict[str, Any]]]] = Field( + default_factory=list, description="List of identifier tuples" + ) + + +class CreateScheduleInput(BaseModel): + professional: int = Field(gt=0, description="Professional ID") + description: str = Field( + min_length=1, max_length=500, description="Appointment description" + ) + customer: int = Field(gt=0, description="Customer ID") + date: str = Field(description="Date in ISO format (e.g., 2026-03-15T10:00:00)") + service_center: int = Field(gt=0, description="Service center ID") + + @field_validator("date") + @classmethod + def validate_date(cls, v: str) -> str: + try: + datetime.fromisoformat(v) + except ValueError: + raise ValueError( + f"Invalid date format: '{v}'. Expected ISO format (e.g., '2026-03-15T10:00:00' or '2026-03-15T10:00:00+00:00')" + ) + return v + + +class FindServiceCentersInput(BaseModel): + offset: int = Field(default=0, ge=0, description="Offset for pagination") + limit: Optional[int] = Field(default=None, ge=1, description="Limit for pagination") + + +class FindProductsInput(BaseModel): + offset: int = Field(default=0, ge=0, description="Offset for pagination") + limit: Optional[int] = Field(default=None, ge=1, description="Limit for pagination") @mcp.tool() def find_customer_by_identifier(identifier: str) -> List[Dict]: - result = tryton_call( - "model.party.identifier.search_read", - [ - [["code", "=", identifier]], - 0, - None, - None, - ["party", "party.name"], - {"company": 1}, - ], - ) + FindCustomerInput(identifier=identifier) - if not result["success"]: + services = get_service_provider() + result = services.party.find_by_identifier(identifier) + + if result.is_error: raise Exception(f"Customer with identifier {identifier} not found.") - return result["data"] + return result.data @mcp.tool() def create_customer( name: str, identifiers: List[tuple[str, List[Dict[str, Any]]]] ) -> List[int]: + CreateCustomerInput(name=name, identifiers=identifiers) - result = tryton_call( - "model.party.party.create", - [{"name": name, "identifiers": identifiers}, {}], - ) + services = get_service_provider() + result = services.party.create(name, identifiers) - if not result["success"]: - raise Exception(f"Failed to create {name} as customer") + if result.is_error: + raise Exception(f"Failed to create {name} as customer: {result.error}") - return result["data"] + return result.data @mcp.tool() def create_schedule( professional: int, description: str, customer: int, date: str, service_center: int ) -> List[int]: + CreateScheduleInput( + professional=professional, + description=description, + customer=customer, + date=date, + service_center=service_center, + ) - try: - dt = datetime.fromisoformat(date) - except ValueError: - raise ValueError( - f"Invalid date format: '{date}'. Expected ISO format (e.g., '2026-03-15T10:00:00' or '2026-03-15T10:00:00+00:00')" - ) + services = get_service_provider() + result = services.schedule.create( + professional, description, customer, date, service_center + ) - utc_dt = dt.astimezone(timezone.utc) + if result.is_error: + raise Exception(f"Failed to create schedule: {result.error}") - example_schedule = [ - { - "professional": professional, - "description": description, - "customer": customer, - "date": utc_dt, - "service_center": service_center, - } - ] - - result = tryton_call("model.naliia.schedule.create", [example_schedule, {}]) - - if not result["success"]: - raise Exception(f"Failed to create schedule: {result['error']}") - - return result["data"] + return result.data @mcp.tool() -def find_service_centers(): - result = tryton_call( - "model.naliia.service_center.search_read", - [[[]], 0, None, None, ["name", "address.street"], {}], - ) +def find_service_centers(offset: int = 0, limit: Optional[int] = None) -> List[Dict]: + FindServiceCentersInput(offset=offset, limit=limit) - if not result["success"]: - raise Exception(f"Failed to find service centers: {result['error']}") + services = get_service_provider() + result = services.service_center.find_all(offset=offset, limit=limit) - return result["data"] + if result.is_error: + raise Exception(f"Failed to find service centers: {result.error}") + + return result.data @mcp.tool() -def find_products_and_services(): - result = tryton_call( - "model.product.product.search_read", - [ - [["active", "=", True], ["salable", "=", True]], - 0, - None, - None, - ["name", "list_price", "description"], - {"company": 1}, - ], - ) +def find_products_and_services( + offset: int = 0, limit: Optional[int] = None +) -> List[Dict]: + FindProductsInput(offset=offset, limit=limit) - if not result["success"]: - raise Exception(f"Failed to find products: {result['error']}") + services = get_service_provider() + result = services.product.find_salable(offset=offset, limit=limit) - return result["data"] + if result.is_error: + raise Exception(f"Failed to find products: {result.error}") + + return result.data diff --git a/src/tryton_mcp/services/__init__.py b/src/tryton_mcp/services/__init__.py new file mode 100644 index 0000000..ef46b48 --- /dev/null +++ b/src/tryton_mcp/services/__init__.py @@ -0,0 +1,14 @@ +from tryton_mcp.services.base import BaseService, TrytonResponse +from tryton_mcp.services.party import PartyService +from tryton_mcp.services.schedule import ScheduleService +from tryton_mcp.services.product import ProductService +from tryton_mcp.services.service_center import ServiceCenterService + +__all__ = [ + "BaseService", + "TrytonResponse", + "PartyService", + "ScheduleService", + "ProductService", + "ServiceCenterService", +] diff --git a/src/tryton_mcp/services/base.py b/src/tryton_mcp/services/base.py new file mode 100644 index 0000000..a14d007 --- /dev/null +++ b/src/tryton_mcp/services/base.py @@ -0,0 +1,28 @@ +from typing import Any, Dict, List, Optional +from dataclasses import dataclass + + +@dataclass +class TrytonResponse: + success: bool + data: Any = None + error: Optional[str] = None + + @property + def is_error(self) -> bool: + return not self.success + + +class BaseService: + def __init__(self, client: Any): + self._client = client + + def call(self, method: str, args: List[Any]) -> TrytonResponse: + try: + response = self._client.call(method, args) + return TrytonResponse(success=True, data=response) + except Exception as e: + return TrytonResponse(success=False, error=str(e)) + + def _get_client(self) -> Any: + return self._client diff --git a/src/tryton_mcp/services/party.py b/src/tryton_mcp/services/party.py new file mode 100644 index 0000000..cc310cb --- /dev/null +++ b/src/tryton_mcp/services/party.py @@ -0,0 +1,43 @@ +from typing import Any, Dict, List, Optional + +from tryton_mcp.services.base import BaseService, TrytonResponse + + +class PartyService(BaseService): + def find_by_identifier(self, identifier: str) -> TrytonResponse: + return self.call( + "model.party.identifier.search_read", + [ + [["code", "=", identifier]], + 0, + None, + None, + ["party", "party.name"], + {"company": 1}, + ], + ) + + def create( + self, name: str, identifiers: List[tuple[str, List[Dict[str, Any]]]] + ) -> TrytonResponse: + return self.call( + "model.party.party.create", + [{"name": name, "identifiers": identifiers}, {}], + ) + + def search_parties( + self, + domain: Optional[List[Any]] = None, + offset: int = 0, + limit: Optional[int] = None, + fields: Optional[List[str]] = None, + ) -> TrytonResponse: + if domain is None: + domain = [] + if fields is None: + fields = ["name", "id"] + + return self.call( + "model.party.party.search_read", + [domain, offset, limit, None, fields, {}], + ) diff --git a/src/tryton_mcp/services/product.py b/src/tryton_mcp/services/product.py new file mode 100644 index 0000000..1a1599b --- /dev/null +++ b/src/tryton_mcp/services/product.py @@ -0,0 +1,39 @@ +from typing import Any, List, Optional + +from tryton_mcp.services.base import BaseService, TrytonResponse + + +class ProductService(BaseService): + def find_salable( + self, + offset: int = 0, + limit: Optional[int] = None, + ) -> TrytonResponse: + return self.call( + "model.product.product.search_read", + [ + [["active", "=", True], ["salable", "=", True]], + offset, + limit, + None, + ["name", "list_price", "description"], + {"company": 1}, + ], + ) + + def search( + self, + domain: Optional[List[Any]] = None, + offset: int = 0, + limit: Optional[int] = None, + fields: Optional[List[str]] = None, + ) -> TrytonResponse: + if domain is None: + domain = [] + if fields is None: + fields = ["id", "name", "list_price"] + + return self.call( + "model.product.product.search_read", + [domain, offset, limit, None, fields, {}], + ) diff --git a/src/tryton_mcp/services/provider.py b/src/tryton_mcp/services/provider.py new file mode 100644 index 0000000..01a971a --- /dev/null +++ b/src/tryton_mcp/services/provider.py @@ -0,0 +1,37 @@ +from tryton_mcp.services.party import PartyService +from tryton_mcp.services.schedule import ScheduleService +from tryton_mcp.services.product import ProductService +from tryton_mcp.services.service_center import ServiceCenterService + + +class ServiceProvider: + def __init__(self, client): + self._client = client + self._party_service: PartyService | None = None + self._schedule_service: ScheduleService | None = None + self._product_service: ProductService | None = None + self._service_center_service: ServiceCenterService | None = None + + @property + def party(self) -> PartyService: + if self._party_service is None: + self._party_service = PartyService(self._client) + return self._party_service + + @property + def schedule(self) -> ScheduleService: + if self._schedule_service is None: + self._schedule_service = ScheduleService(self._client) + return self._schedule_service + + @property + def product(self) -> ProductService: + if self._product_service is None: + self._product_service = ProductService(self._client) + return self._product_service + + @property + def service_center(self) -> ServiceCenterService: + if self._service_center_service is None: + self._service_center_service = ServiceCenterService(self._client) + return self._service_center_service diff --git a/src/tryton_mcp/services/schedule.py b/src/tryton_mcp/services/schedule.py new file mode 100644 index 0000000..7d45f42 --- /dev/null +++ b/src/tryton_mcp/services/schedule.py @@ -0,0 +1,53 @@ +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional + +from tryton_mcp.services.base import BaseService, TrytonResponse + + +class ScheduleService(BaseService): + def create( + self, + professional: int, + description: str, + customer: int, + date: str, + service_center: int, + ) -> TrytonResponse: + try: + dt = datetime.fromisoformat(date) + except ValueError as e: + return TrytonResponse( + success=False, + error=f"Invalid date format: '{date}'. Expected ISO format. Error: {e}", + ) + + utc_dt = dt.astimezone(timezone.utc) + + schedule_data = [ + { + "professional": professional, + "description": description, + "customer": customer, + "date": utc_dt, + "service_center": service_center, + } + ] + + return self.call("model.naliia.schedule.create", [schedule_data, {}]) + + def search( + self, + domain: Optional[List[Any]] = None, + offset: int = 0, + limit: Optional[int] = None, + fields: Optional[List[str]] = None, + ) -> TrytonResponse: + if domain is None: + domain = [] + if fields is None: + fields = ["id", "professional", "description", "customer", "date"] + + return self.call( + "model.naliia.schedule.search_read", + [domain, offset, limit, None, fields, {}], + ) diff --git a/src/tryton_mcp/services/service_center.py b/src/tryton_mcp/services/service_center.py new file mode 100644 index 0000000..506fbee --- /dev/null +++ b/src/tryton_mcp/services/service_center.py @@ -0,0 +1,32 @@ +from typing import Any, List, Optional + +from tryton_mcp.services.base import BaseService, TrytonResponse + + +class ServiceCenterService(BaseService): + def find_all( + self, + offset: int = 0, + limit: Optional[int] = None, + ) -> TrytonResponse: + return self.call( + "model.naliia.service_center.search_read", + [[[]], offset, limit, None, ["name", "address.street"], {}], + ) + + def search( + self, + domain: Optional[List[Any]] = None, + offset: int = 0, + limit: Optional[int] = None, + fields: Optional[List[str]] = None, + ) -> TrytonResponse: + if domain is None: + domain = [] + if fields is None: + fields = ["id", "name", "address.street"] + + return self.call( + "model.naliia.service_center.search_read", + [domain, offset, limit, None, fields, {}], + ) diff --git a/tests/test_customer.py b/tests/test_customer.py index cafa629..7cb30e1 100644 --- a/tests/test_customer.py +++ b/tests/test_customer.py @@ -1,48 +1,61 @@ import pytest -import sys -from pathlib import Path +import pytest_asyncio +from unittest.mock import MagicMock, patch import json -from tryton_mcp.server import create_customer - +import sys +from pathlib import Path sys.path.insert(0, str(Path(__file__).parent / "src")) +@pytest.fixture +def mock_client(): + return MagicMock() + + +@pytest.fixture +def mock_service_provider(mock_client): + provider_mock = MagicMock() + mock_response = MagicMock() + mock_response.is_error = False + mock_response.data = [] + + provider_mock.party.find_by_identifier.return_value = mock_response + provider_mock.party.create.return_value = mock_response + + return provider_mock + + +@pytest.fixture +def server_with_mocks(mock_service_provider): + with patch("tryton_mcp.server._service_provider", mock_service_provider): + yield mock_service_provider + + @pytest.mark.asyncio class TestParty: + async def test_create_new_customer(self, server_with_mocks): + from tryton_mcp.server import create_customer - async def test_create_new_customer(self, mcp_client, mock_settings): - mock_client = mock_settings.get_client.return_value - mock_client.call.return_value = [1] + server_with_mocks.party.create.return_value.data = [1] - result = await mcp_client.call_tool( - "create_customer", - { - "name": "Alejandro Zapata", - "identifiers": [["create", [{"type": "mobile", "code": "310659595"}]]], - }, + result = create_customer( + name="Alejandro Zapata", + identifiers=[["create", [{"type": "mobile", "code": "310659595"}]]], ) - assert result.content is not None - assert len(result.content) == 1 - assert result.content[0].text == "[1]" + assert result == [1] - async def test_find_customer(self, mcp_client, mock_settings): + async def test_find_customer(self, server_with_mocks): + from tryton_mcp.server import find_customer_by_identifier expected_return = [ - {'id': 19, 'party': 44, 'party.': {'name': 'Alejandro Zapata', 'id': 44}} + {"id": 19, "party": 44, "party.": {"name": "Alejandro Zapata", "id": 44}} ] - mock_client = mock_settings.get_client.return_value - mock_client.call.return_value = expected_return - - result = await mcp_client.call_tool( - "find_customer_by_identifier", - { - "identifier": "310659595" - }) - - assert result.content is not None - assert len(result.content) == 1 - assert json.loads(result.content[0].text) == expected_return + server_with_mocks.party.find_by_identifier.return_value.data = expected_return + + result = find_customer_by_identifier("310659595") + + assert result == expected_return diff --git a/tests/test_server.py b/tests/test_server.py index f477f52..b4e7263 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -1,8 +1,8 @@ import pytest +import pytest_asyncio from unittest.mock import MagicMock, patch -import datetime import json -from fastmcp.exceptions import ToolError +from contextlib import asynccontextmanager import sys from pathlib import Path @@ -10,146 +10,199 @@ from pathlib import Path sys.path.insert(0, str(Path(__file__).parent / "src")) +@pytest.fixture +def mock_client(): + return MagicMock() + + +@pytest.fixture +def mock_service_provider(mock_client): + provider_mock = MagicMock() + mock_response = MagicMock() + mock_response.is_error = False + mock_response.data = [] + + provider_mock.party.find_by_identifier.return_value = mock_response + provider_mock.party.create.return_value = mock_response + provider_mock.schedule.create.return_value = mock_response + provider_mock.service_center.find_all.return_value = mock_response + provider_mock.product.find_salable.return_value = mock_response + + return provider_mock + + +@pytest.fixture +def server_with_mocks(mock_service_provider): + with patch("tryton_mcp.server._service_provider", mock_service_provider): + yield mock_service_provider + + @pytest.mark.asyncio -class TestTrytonCall: - async def test_tryton_call_success(self, mock_settings): - from tryton_mcp.server import tryton_call +class TestFindCustomerByIdentifier: + async def test_find_customer_success(self, server_with_mocks): + from tryton_mcp.server import find_customer_by_identifier - mock_client = mock_settings.get_client.return_value - mock_client.call.return_value = [{"id": 1, "name": "Test"}] + server_with_mocks.party.find_by_identifier.return_value.data = [ + {"id": 1, "party": {"name": "Test Customer"}} + ] - result = tryton_call("model.test.read", [[1], ["name"]]) + result = find_customer_by_identifier("12345") - assert result["success"] is True - assert result["data"] == [{"id": 1, "name": "Test"}] - mock_client.call.assert_called_once_with("model.test.read", [[1], ["name"]]) + assert result == [{"id": 1, "party": {"name": "Test Customer"}}] + server_with_mocks.party.find_by_identifier.assert_called_once_with("12345") - async def test_tryton_call_failure(self, mock_settings): - from tryton_mcp.server import tryton_call + async def test_find_customer_not_found(self, server_with_mocks): + from tryton_mcp.server import find_customer_by_identifier - mock_client = mock_settings.get_client.return_value - mock_client.call.side_effect = Exception("Connection error") + server_with_mocks.party.find_by_identifier.return_value.is_error = True - result = tryton_call("model.test.read", [[1], ["name"]]) + with pytest.raises(Exception, match="Customer with identifier"): + find_customer_by_identifier("invalid") - assert result["success"] is False - assert "Connection error" in result["error"] + +@pytest.mark.asyncio +class TestCreateCustomer: + async def test_create_customer_success(self, server_with_mocks): + from tryton_mcp.server import create_customer + + server_with_mocks.party.create.return_value.data = [1] + + result = create_customer("New Customer", []) + + assert result == [1] + + async def test_create_customer_invalid_name(self, server_with_mocks): + from tryton_mcp.server import create_customer + from pydantic import ValidationError + + with pytest.raises(ValidationError): + create_customer("", []) @pytest.mark.asyncio class TestCreateSchedule: - async def test_create_schedule_success(self, mcp_client, mock_settings): + async def test_create_schedule_success(self, server_with_mocks): from tryton_mcp.server import create_schedule - mock_client = mock_settings.get_client.return_value - mock_client.call.return_value = [1] + server_with_mocks.schedule.create.return_value.data = [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, - }, + result = 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]" + assert result == [1] - mock_client.call.assert_called_once() - - async def test_create_schedule_failure(self, mcp_client, mock_settings): + async def test_create_schedule_failure(self, server_with_mocks): from tryton_mcp.server import create_schedule - mock_client = mock_settings.get_client.return_value - mock_client.call.side_effect = Exception("Create failed") + server_with_mocks.schedule.create.return_value.is_error = True + server_with_mocks.schedule.create.return_value.error = "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, - }, + create_schedule( + professional=6, + description="Test appointment", + customer=4, + date="2026-03-15T10:00:00", + service_center=11, ) - async def test_create_schedule_invalid_date_format(self, mcp_client, mock_settings): - with pytest.raises(ToolError, match="Invalid date format"): - await mcp_client.call_tool( - "create_schedule", - { - "professional": 6, - "description": "Test appointment", - "customer": 4, - "date": "invalid-date", - "service_center": 11, - }, + async def test_create_schedule_invalid_date_format(self, server_with_mocks): + from tryton_mcp.server import create_schedule + from pydantic import ValidationError + + with pytest.raises(ValidationError): + create_schedule( + professional=6, + description="Test appointment", + customer=4, + date="invalid-date", + service_center=11, + ) + + async def test_create_schedule_invalid_professional(self, server_with_mocks): + from tryton_mcp.server import create_schedule + from pydantic import ValidationError + + with pytest.raises(ValidationError): + create_schedule( + professional=0, + 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): + async def test_find_service_centers_success(self, server_with_mocks): from tryton_mcp.server import find_service_centers - mock_client = mock_settings.get_client.return_value - mock_client.call.return_value = [ + server_with_mocks.service_center.find_all.return_value.data = [ {"id": 11, "name": "Center 1", "address.street": "Street 1"} ] - result = await mcp_client.call_tool("find_service_centers", {}) + result = 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] == [[]] + assert result == [{"id": 11, "name": "Center 1", "address.street": "Street 1"}] + server_with_mocks.service_center.find_all.assert_called_once_with( + offset=0, limit=None + ) - async def test_find_service_centers_failure(self, mcp_client, mock_settings): + async def test_find_service_centers_with_pagination(self, server_with_mocks): from tryton_mcp.server import find_service_centers - mock_client = mock_settings.get_client.return_value - mock_client.call.side_effect = Exception("Search failed") + find_service_centers(offset=10, limit=5) + + server_with_mocks.service_center.find_all.assert_called_once_with( + offset=10, limit=5 + ) + + async def test_find_service_centers_failure(self, server_with_mocks): + from tryton_mcp.server import find_service_centers + + server_with_mocks.service_center.find_all.return_value.is_error = True + server_with_mocks.service_center.find_all.return_value.error = "Search failed" with pytest.raises(Exception, match="Failed to find service centers"): - await mcp_client.call_tool("find_service_centers", {}) + find_service_centers() @pytest.mark.asyncio class TestFindProductsAndServices: - async def test_find_products_and_services_success(self, mcp_client, mock_settings): + async def test_find_products_and_services_success(self, server_with_mocks): from tryton_mcp.server import find_products_and_services - mock_client = mock_settings.get_client.return_value - mock_client.call.return_value = [ + server_with_mocks.product.find_salable.return_value.data = [ {"id": 1, "name": "Product 1", "list_price": 100.0} ] - result = await mcp_client.call_tool("find_products_and_services", {}) + result = find_products_and_services() - result_data = json.loads(result.content[0].text) - assert result_data == [{"id": 1, "name": "Product 1", "list_price": 100.0}] + assert result == [{"id": 1, "name": "Product 1", "list_price": 100.0}] + server_with_mocks.product.find_salable.assert_called_once_with( + offset=0, limit=None + ) - 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): + async def test_find_products_with_pagination(self, server_with_mocks): 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") + find_products_and_services(offset=5, limit=10) + + server_with_mocks.product.find_salable.assert_called_once_with( + offset=5, limit=10 + ) + + async def test_find_products_failure(self, server_with_mocks): + from tryton_mcp.server import find_products_and_services + + server_with_mocks.product.find_salable.return_value.is_error = True + server_with_mocks.product.find_salable.return_value.error = "Search failed" with pytest.raises(Exception, match="Failed to find products"): - await mcp_client.call_tool("find_products_and_services", {}) + find_products_and_services()