maskanx_cm_backend/tests/test_crm_properties_mcp.py

# -*- coding: utf-8 -*-
"""Creating CRM fields from chat, through MCP.

This is the path the user described: ask in MaskanX chat for a new field,
and it appears in the CRM. The tests cover the client calls and the MCP
tool dispatch — the CRM's own validation is tested in the CRM's suite.
"""
from __future__ import annotations

import json

from adclaw.integrations import maskan_crm
from adclaw.integrations.maskan_crm import MaskanCRMClient
from adclaw.tools import maskan_crm_mcp


class _Response:
    def __init__(self, payload) -> None:
        self.payload = payload

    def __enter__(self):
        return self

    def __exit__(self, *_args):
        return False

    def read(self) -> bytes:
        return json.dumps(self.payload).encode("utf-8")


def _client() -> MaskanCRMClient:
    return MaskanCRMClient("http://crm.internal:8091", "mcrm_test-key")


# --- the client ---


def test_listing_properties_accepts_the_arrays_the_api_returns(monkeypatch):
    """Every other endpoint returns an object; collections return a list."""
    monkeypatch.setattr(
        maskan_crm,
        "urlopen",
        lambda request, timeout: _Response(
            [{"name": "budget_range", "data_type": "enumeration"}],
        ),
    )

    result = _client().list_properties()

    assert result["properties"][0]["name"] == "budget_range"


def test_listing_can_be_narrowed_to_one_object_type(monkeypatch):
    captured = {}

    def fake_urlopen(request, timeout):
        captured["url"] = request.full_url
        return _Response([])

    monkeypatch.setattr(maskan_crm, "urlopen", fake_urlopen)

    _client().list_properties("contact")

    assert captured["url"].endswith("/integrations/properties?object_type=contact")


def test_creating_a_property_posts_the_definition(monkeypatch):
    captured = {}

    def fake_urlopen(request, timeout):
        captured["url"] = request.full_url
        captured["method"] = request.get_method()
        captured["body"] = json.loads(request.data.decode("utf-8"))
        return _Response({"id": "prop_1", "name": "budget_range"})

    monkeypatch.setattr(maskan_crm, "urlopen", fake_urlopen)

    result = _client().create_property(
        {
            "object_type": "contact",
            "name": "budget_range",
            "label": "Budget range",
            "data_type": "enumeration",
            "options": ["50-70 lakh"],
        },
    )

    assert result["id"] == "prop_1"
    assert captured["method"] == "POST"
    assert captured["url"].endswith("/integrations/properties")
    assert captured["body"]["options"] == ["50-70 lakh"]


def test_a_non_list_non_object_response_is_still_refused(monkeypatch):
    """An unexpected scalar is more likely a proxy page than data."""
    monkeypatch.setattr(
        maskan_crm, "urlopen", lambda request, timeout: _Response("gateway timeout"),
    )

    try:
        _client().list_properties()
    except maskan_crm.MaskanCRMRequestError as exc:
        assert "unexpected response" in str(exc)
    else:  # pragma: no cover - the call must not succeed
        raise AssertionError("a scalar response should have been refused")


# --- the MCP tools ---


def _call(tool_name, arguments=None):
    return maskan_crm_mcp._handle_request(
        {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/call",
            "params": {"name": tool_name, "arguments": arguments or {}},
        },
    )


def _payload(response):
    return json.loads(response["result"]["content"][0]["text"])


def test_both_property_tools_are_advertised():
    listed = maskan_crm_mcp._handle_request(
        {"jsonrpc": "2.0", "id": 1, "method": "tools/list"},
    )

    names = {tool["name"] for tool in listed["result"]["tools"]}
    assert {"list_maskan_crm_properties", "create_maskan_crm_property"} <= names


def test_the_create_tool_tells_the_agent_to_look_before_adding():
    """Otherwise it invents a near-duplicate of a field that already exists."""
    listed = maskan_crm_mcp._handle_request(
        {"jsonrpc": "2.0", "id": 1, "method": "tools/list"},
    )
    tool = next(
        t
        for t in listed["result"]["tools"]
        if t["name"] == "create_maskan_crm_property"
    )

    assert "list_maskan_crm_properties" in tool["description"]
    assert set(tool["inputSchema"]["required"]) == {
        "object_type", "name", "label", "data_type",
    }


def test_listing_properties_through_mcp(monkeypatch):
    monkeypatch.setattr(
        maskan_crm_mcp,
        "configured_maskan_crm_client",
        lambda: type(
            "C", (), {"list_properties": lambda self, ot=None: {"properties": []}},
        )(),
    )

    response = _call("list_maskan_crm_properties", {"object_type": "contact"})

    assert response["result"]["isError"] is False
    assert _payload(response) == {"properties": []}


def test_creating_a_property_through_mcp(monkeypatch):
    captured = {}

    class _Client:
        def create_property(self, payload):
            captured["payload"] = payload
            return {"id": "prop_1", "name": "budget_range"}

    monkeypatch.setattr(
        maskan_crm_mcp, "configured_maskan_crm_client", lambda: _Client(),
    )

    response = _call(
        "create_maskan_crm_property",
        {
            "object_type": "contact",
            "name": "budget_range",
            "label": "Budget range",
            "data_type": "enumeration",
            "options": ["50-70 lakh"],
            "description": None,
        },
    )

    assert response["result"]["isError"] is False
    assert _payload(response)["id"] == "prop_1"
    # Nulls are stripped so the CRM's defaults apply rather than being
    # overwritten with None.
    assert "description" not in captured["payload"]


def test_a_rejected_definition_comes_back_as_a_readable_error(monkeypatch):
    """An agent needs to read why, so it can correct itself and retry."""

    class _Client:
        def create_property(self, payload):
            raise maskan_crm.MaskanCRMRequestError(
                "'Budget Range' is not a usable property name.",
                status_code=422,
            )

    monkeypatch.setattr(
        maskan_crm_mcp, "configured_maskan_crm_client", lambda: _Client(),
    )

    response = _call(
        "create_maskan_crm_property",
        {
            "object_type": "contact",
            "name": "Budget Range",
            "label": "Budget range",
            "data_type": "string",
        },
    )

    assert response["result"]["isError"] is True
    assert "usable property name" in _payload(response)["error"]