refactor: add service layer architecture with Pydantic validation
- Add services layer (PartyService, ScheduleService, ProductService, ServiceCenterService) - Add ServiceProvider for dependency injection - Add Pydantic input validation on all MCP tools - Add thread-safety to TrytonSettings with double-checked locking - Update tests to mock ServiceProvider instead of direct client calls - Add pydantic dependency
This commit is contained in:
@@ -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]
|
||||
|
||||
@@ -0,0 +1,4 @@
|
||||
from tryton_mcp.config import settings, TrytonSettings
|
||||
from tryton_mcp.server import mcp
|
||||
|
||||
__all__ = ["settings", "TrytonSettings", "mcp"]
|
||||
|
||||
@@ -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 {
|
||||
@@ -38,6 +42,8 @@ class TrytonSettings:
|
||||
}
|
||||
|
||||
def get_client(self) -> Client:
|
||||
if self._client is None:
|
||||
with self._lock:
|
||||
if self._client is None:
|
||||
self._client = Client(**self.to_dict())
|
||||
return self._client
|
||||
|
||||
@@ -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]:
|
||||
|
||||
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:
|
||||
client = settings.get_client()
|
||||
response = client.call(name, args)
|
||||
return {"success": True, "data": response}
|
||||
except Exception as e:
|
||||
return {"success": False, "error": str(e)}
|
||||
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]:
|
||||
|
||||
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')"
|
||||
CreateScheduleInput(
|
||||
professional=professional,
|
||||
description=description,
|
||||
customer=customer,
|
||||
date=date,
|
||||
service_center=service_center,
|
||||
)
|
||||
|
||||
utc_dt = dt.astimezone(timezone.utc)
|
||||
services = get_service_provider()
|
||||
result = services.schedule.create(
|
||||
professional, description, customer, date, service_center
|
||||
)
|
||||
|
||||
example_schedule = [
|
||||
{
|
||||
"professional": professional,
|
||||
"description": description,
|
||||
"customer": customer,
|
||||
"date": utc_dt,
|
||||
"service_center": service_center,
|
||||
}
|
||||
]
|
||||
if result.is_error:
|
||||
raise Exception(f"Failed to create schedule: {result.error}")
|
||||
|
||||
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
|
||||
|
||||
14
src/tryton_mcp/services/__init__.py
Normal file
14
src/tryton_mcp/services/__init__.py
Normal file
@@ -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",
|
||||
]
|
||||
28
src/tryton_mcp/services/base.py
Normal file
28
src/tryton_mcp/services/base.py
Normal file
@@ -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
|
||||
43
src/tryton_mcp/services/party.py
Normal file
43
src/tryton_mcp/services/party.py
Normal file
@@ -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, {}],
|
||||
)
|
||||
39
src/tryton_mcp/services/product.py
Normal file
39
src/tryton_mcp/services/product.py
Normal file
@@ -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, {}],
|
||||
)
|
||||
37
src/tryton_mcp/services/provider.py
Normal file
37
src/tryton_mcp/services/provider.py
Normal file
@@ -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
|
||||
53
src/tryton_mcp/services/schedule.py
Normal file
53
src/tryton_mcp/services/schedule.py
Normal file
@@ -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, {}],
|
||||
)
|
||||
32
src/tryton_mcp/services/service_center.py
Normal file
32
src/tryton_mcp/services/service_center.py
Normal file
@@ -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, {}],
|
||||
)
|
||||
@@ -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
|
||||
server_with_mocks.party.find_by_identifier.return_value.data = expected_return
|
||||
|
||||
result = await mcp_client.call_tool(
|
||||
"find_customer_by_identifier",
|
||||
{
|
||||
"identifier": "310659595"
|
||||
})
|
||||
result = find_customer_by_identifier("310659595")
|
||||
|
||||
assert result.content is not None
|
||||
assert len(result.content) == 1
|
||||
assert json.loads(result.content[0].text) == expected_return
|
||||
assert result == expected_return
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user