RollCalcPython/tests/test_mcp_adapter.py

import sys
import types
import unittest
from unittest.mock import patch

import mcp_server
from roll_calculation import calculate_roll, get_article
from rollcalc_mcp_tools import (
    analyze_transport_capacity_result,
    calculate_roll_diameter_result,
    get_article_result,
)
from transport_calculation import analyze_transport, transport_presets


class _FakeFastMCP:
    def __init__(self, name):
        self.name = name
        self.tools = []

    def tool(self):
        def register(function):
            self.tools.append(function)
            return function
        return register


class McpAdapterTests(unittest.TestCase):
    def test_get_article_delegates_to_the_domain_for_success_not_found_and_conflict(self):
        for arguments in (
            {"article_number": "146900"},
            {"article_number": "does-not-exist"},
            {"article_number": "146900", "article_name_hint": "Other product"},
        ):
            with self.subTest(arguments=arguments):
                self.assertEqual(
                    get_article_result(**arguments),
                    get_article(**arguments),
                )

    def test_calculate_roll_diameter_delegates_for_normal_weighted_and_invalid_cases(self):
        for arguments in (
            {
                "article_number": "146900",
                "roll_length_m": 50.0,
                "core_diameter_mm": 150.0,
            },
            {
                "article_number": "146900",
                "roll_length_m": 50.0,
                "core_diameter_mm": 150.0,
                "width_m": 5.8,
                "area_weight_g_m2": 1767.2,
            },
            {"roll_length_m": -1.0, "core_diameter_mm": 150.0, "thickness_mm": 2.0},
        ):
            with self.subTest(arguments=arguments):
                actual = calculate_roll_diameter_result(**arguments)
                expected = calculate_roll({**arguments, "include_roll_weight": True})
                if expected["status"] == "success":
                    self.assertEqual(
                        actual["transport_roll_inputs"]["roll_diameter_mm"],
                        845.9,
                    )
                    del actual["transport_roll_inputs"]
                self.assertEqual(actual, expected)

    def test_roll_calculation_adapter_always_requests_weight_from_the_domain(self):
        result = calculate_roll_diameter_result(
            article_number="114030",
            roll_length_m=15.0,
            core_diameter_mm=100.0,
        )

        self.assertEqual(result["status"], "success")
        self.assertIsNotNone(result["calculation"]["roll_weight_kg"])

    def test_roll_calculation_adapter_uses_shared_transport_input_helper(self):
        expected_inputs = {"roll_diameter_mm": 123.4}
        with patch(
            "rollcalc_mcp_tools.transport_roll_inputs_from_roll_calculation",
            return_value=expected_inputs,
        ) as helper:
            result = calculate_roll_diameter_result(
                article_number="114030",
                roll_length_m=15.0,
                core_diameter_mm=100.0,
            )

        helper.assert_called_once()
        self.assertEqual(result["transport_roll_inputs"], expected_inputs)

    def test_transport_tool_delegates_for_preset_custom_and_invalid_cases(self):
        for arguments in (
            {
                "transport_preset": "container_20ft",
                "roll_diameter_mm": 1000.0,
                "core_diameter_mm": 150.0,
                "roll_width_m": 2.0,
                "roll_weight_kg": 1000.0,
                "product_length_m": 50.0,
            },
            {
                "length_m": 10.0,
                "width_m": 5.0,
                "height_m": 5.0,
                "max_weight_kg": 10000.0,
                "roll_diameter_mm": 1000.0,
                "core_diameter_mm": 150.0,
                "roll_width_m": 1.0,
                "roll_weight_kg": 1000.0,
            },
            {"length_m": 0.0},
        ):
            with self.subTest(arguments=arguments):
                self.assertEqual(
                    analyze_transport_capacity_result(**arguments),
                    analyze_transport(arguments),
                )

    def test_server_registers_exactly_the_three_stdio_poc_tools(self):
        mcp = types.ModuleType("mcp")
        server = types.ModuleType("mcp.server")
        fastmcp = types.ModuleType("mcp.server.fastmcp")
        fastmcp.FastMCP = _FakeFastMCP
        with patch.dict(sys.modules, {
            "mcp": mcp,
            "mcp.server": server,
            "mcp.server.fastmcp": fastmcp,
        }):
            instance = mcp_server.create_server()

        self.assertEqual(instance.name, "RollCalc")
        self.assertEqual(
            [tool.__name__ for tool in instance.tools],
            [
                "get_article",
                "calculate_roll_diameter",
                "analyze_transport_capacity",
            ],
        )

    def test_server_schema_exposes_weight_and_transport_chaining_contract(self):
        instance = mcp_server.create_server()
        tools = instance._tool_manager._tools
        roll_schema = tools["calculate_roll_diameter"].parameters
        transport_tool = tools["analyze_transport_capacity"]
        transport_schema = transport_tool.parameters

        self.assertNotIn("include_roll_weight", roll_schema["properties"])
        preset_schema = transport_schema["properties"]["transport_preset"]
        self.assertEqual(
            preset_schema["anyOf"][0]["enum"],
            [preset.key for preset in transport_presets()],
        )
        self.assertEqual(
            transport_schema["required"],
            [
                "roll_diameter_mm",
                "core_diameter_mm",
                "roll_width_m",
                "roll_weight_kg",
                "product_length_m",
            ],
        )
        for field in transport_schema["required"]:
            self.assertEqual(
                transport_schema["properties"][field]["type"], "number"
            )
        description = transport_tool.description
        for text in (
            "first call get_article and calculate_roll_diameter",
            "transport_roll_inputs bundle unchanged",
            "Do not choose among minimum, average, or maximum diameter",
            "product_length_m",
            "stateless tool does not recalculate a roll",
        ):
            self.assertIn(text, description)


if __name__ == "__main__":
    unittest.main()