mcp/tests/test_arrange_destination_trailer_tool.py

import unittest

from app import GatewayApp
from public_gateway import PublicGatewayApp
from services.output_presenter import OutputPresenter
from tools.arrange_destination_trailer import ArrangeDestinationTrailerTool


class RecordingApiClient:
    def __init__(self):
        self.calls = []

    def call_tool(self, name, route_path, payload, request_id):
        self.calls.append((name, route_path, payload, request_id))
        return {'code': 'MCP_0000', 'data': {}, 'meta': {'request_id': request_id}}


class ArrangeDestinationTrailerToolTest(unittest.TestCase):
    def test_metadata_rejects_unknown_fields(self):
        schema = ArrangeDestinationTrailerTool().metadata()['input_schema']
        self.assertFalse(schema['additionalProperties'])
        self.assertEqual(['trailer_provider'], schema['required'])
        self.assertNotIn('outbound_id', schema['properties'])

    def test_call_requires_a_number_and_posts_the_route(self):
        client = RecordingApiClient()
        tool = ArrangeDestinationTrailerTool(api_client=client)
        with self.assertRaises(ValueError):
            tool.call(trailer_provider='甲拖车-A')
        tool.call(trailer_provider='甲拖车-A', container_code='CONT1')
        self.assertEqual('arrange_destination_trailer', client.calls[0][0])
        self.assertEqual('/mcp/tools/arrangeDestinationTrailer', client.calls[0][1])
        self.assertEqual(
            {'trailer_provider': '甲拖车-A', 'container_code': 'CONT1'},
            client.calls[0][2],
        )

    def test_presenter_returns_only_three_labels(self):
        presenter = OutputPresenter()
        result = presenter.present('arrange_destination_trailer', {
            'code': 'MCP_0000',
            'data': {
                'trailer_number': 'DET1',
                'bl_number': 'BL1',
                'container_code': 'CONT1',
            },
            'meta': {'request_id': 'rq1'},
        })
        self.assertEqual({
            '拖车单号': 'DET1',
            '提单号': 'BL1',
            '柜号': 'CONT1',
        }, result['structured_content'])
        self.assertFalse(presenter.present('arrange_destination_trailer', {
            'code': 'MCP_0000',
            'data': {
                'trailer_number': 'DET1',
                'bl_number': 'BL1',
                'container_code': 'CONT1',
                'providers_id': 1,
            },
            'meta': {'request_id': 'rq1'},
        })['is_error'] is False)
        self.assertTrue(presenter.present('arrange_destination_trailer', {
            'code': 'MCP_0000',
            'data': {
                'trailer_number': 1,
                'bl_number': 'BL1',
                'container_code': 'CONT1',
            },
            'meta': {'request_id': 'rq1'},
        })['is_error'])

    def test_call_rejects_bad_values_and_forwards_spoken_fields(self):
        tool = ArrangeDestinationTrailerTool()
        with self.assertRaises(RuntimeError):
            tool.call(trailer_provider='甲拖车-A', container_code='CONT1')

        client = RecordingApiClient()
        tool = ArrangeDestinationTrailerTool(api_client=client)
        with self.assertRaises(ValueError):
            tool.call(trailer_provider='  ', container_code='CONT1')
        with self.assertRaises(ValueError):
            tool.call(trailer_provider='甲拖车-A', container_code=1)
        with self.assertRaises(ValueError):
            tool.call(trailer_provider='甲拖车-A', bl_number='BL1', is_direct_send='未知')
        with self.assertRaises(ValueError):
            tool.call(
                trailer_provider='甲拖车-A',
                bl_number='BL1',
                est_pickup_time='2026-10-12',
            )
        with self.assertRaises(ValueError):
            tool.call(
                trailer_provider='甲拖车-A',
                bl_number='BL1',
                order_measures=[{'weight': '1'}],
            )
        with self.assertRaises(ValueError):
            tool.call(
                trailer_provider='甲拖车-A',
                bl_number='BL1',
                order_measures='1',
            )
        with self.assertRaises(ValueError):
            tool.call(
                trailer_provider='甲拖车-A',
                bl_number='BL1',
                order_measures=[1],
            )
        with self.assertRaises(ValueError):
            tool.call(
                trailer_provider='甲拖车-A',
                bl_number='BL1',
                order_measures=[{'order_number': 'SO1'}] * 501,
            )
        with self.assertRaises(ValueError):
            tool.call(trailer_provider='甲' * 201, container_code='CONT1')
        with self.assertRaises(ValueError):
            tool.call(
                trailer_provider='甲拖车-A',
                bl_number='BL1',
                direct_change_reason='其他',
            )

        tool.call(
            trailer_provider=' 甲拖车-A ',
            bl_number=' BL1 ',
            container_code=' CONT1 ',
            is_direct_send='否',
            warehouse_code='LAX1',
            warehouse_contact='张三',
            warehouse_tel='10086',
            follow_up_user='空',
            est_pickup_time='2026-10-12 08:00:00',
            is_direct_change='是',
            direct_change_reason='直送爆仓被拒',
            order_measures=[{
                'order_number': 'SO1',
                'weight': '1.5',
                'volume': None,
            }],
            request_id='rq_case',
        )
        self.assertEqual({
            'trailer_provider': '甲拖车-A',
            'bl_number': 'BL1',
            'container_code': 'CONT1',
            'is_direct_send': '否',
            'warehouse_code': 'LAX1',
            'warehouse_contact': '张三',
            'warehouse_tel': '10086',
            'follow_up_user': '空',
            'est_pickup_time': '2026-10-12 08:00:00',
            'is_direct_change': '是',
            'direct_change_reason': '直送爆仓被拒',
            'order_measures': [{'order_number': 'SO1', 'weight': '1.5'}],
        }, client.calls[-1][2])
        self.assertEqual('rq_case', client.calls[-1][3])

    def test_presenter_keeps_known_arrange_errors(self):
        presenter = OutputPresenter()
        result = presenter.present('arrange_destination_trailer', {
            'code': 'MCP_1401',
            'msg': '请填写订单O1的重量和体积',
            'meta': {'request_id': 'rq1'},
        })
        self.assertEqual('请填写订单O1的重量和体积', result['structured_content']['message'])
        known = presenter.present('arrange_destination_trailer', {
            'code': 'MCP_1401',
            'msg': '安排拖车行失败',
            'meta': {'request_id': 'rq1'},
        })
        self.assertEqual('安排拖车行失败', known['structured_content']['message'])
        self.assertFalse(result['structured_content']['retryable'])
        hidden = presenter.present('arrange_destination_trailer', {
            'code': 'MCP_1401',
            'msg': 'SQLSTATE[42S02]',
            'meta': {'request_id': 'rq1'},
        })
        self.assertEqual('工具参数不正确，请检查后重试', hidden['structured_content']['message'])

    def test_registered_on_both_gateways(self):
        local = GatewayApp().registered_tool_names()
        public = PublicGatewayApp(None, None).registered_tool_names()
        self.assertEqual(local, public)
        self.assertIn('arrange_destination_trailer', local)
        self.assertEqual(51, len(local))
        self.assertIn('arrange_destination_trailer', OutputPresenter.SAFE_TOOLS)


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