mcp/tests/test_output_presenter.py

import json
import unittest

from services.output_presenter import OutputPresenter
from tools.export_out_of_province_port_data import ExportOutOfProvincePortDataTool
from tools.export_pending_outbound_orders import ExportPendingOutboundOrdersTool
from tools.list_order_filter_options import ListOrderFilterOptionsTool
from tools.list_pending_outbound_export_filter_options import (
    ListPendingOutboundExportFilterOptionsTool,
)
from tools.query_order import QueryOrderTool
from tools.query_order_exact import QueryOrderExactTool
from tools.query_track import QueryTrackTool


class OutputPresenterTest(unittest.TestCase):
    def setUp(self):
        self.presenter = OutputPresenter()

    def test_exact_order_uses_labels_and_drops_internal_fields(self):
        result = self.presenter.present(
            'query_order_exact',
            {
                'code': 'MCP_0000',
                'data': {
                    'summary': '当前页返回 1 条订单',
                    'columns': [
                        {'key': 'order_number', 'name': '订单号'},
                        {
                            'key': 'tracking_number',
                            'name': '快递单号',
                            'description': '承运商跟踪号码',
                        },
                    ],
                    'records': [{
                        'order_number': 'SO001',
                        'tracking_number': 'TN001',
                        'order_id': 99,
                        'time_zone': '8.00',
                    }],
                    'tips': [],
                },
                'meta': {
                    'page': 1,
                    'limit': 20,
                    'has_more': False,
                    'request_id': 'rq_exact',
                },
            },
        )

        self.assertFalse(result['is_error'])
        self.assertEqual(
            [
                {'label': '订单号'},
                {'label': '快递单号', 'description': '承运商跟踪号码'},
            ],
            result['structured_content']['headers'],
        )
        self.assertEqual(
            [['SO001', 'TN001']],
            result['structured_content']['rows'],
        )
        self.assertEqual(
            {'page': 1, 'limit': 20, 'has_more': False},
            result['structured_content']['pagination'],
        )
        self.assertEqual({'request_id': 'rq_exact'}, result['meta'])
        serialized = json.dumps(result, ensure_ascii=False)
        self.assertNotIn('order_number', serialized)
        self.assertNotIn('tracking_number', serialized)
        self.assertNotIn('order_id', serialized)
        self.assertNotIn('time_zone', serialized)
        self.assertIn('订单号', result['text'])
        self.assertIn('SO001', result['text'])

    def test_table_missing_value_is_empty_and_duplicate_labels_are_preserved(self):
        result = self.presenter.present(
            'query_track',
            {
                'code': '0',
                'data': {
                    'columns': [
                        {'key': 'status', 'name': '状态'},
                        {'key': 'content', 'name': '状态'},
                    ],
                    'records': [{'status': '已发货'}],
                    'tips': ['仅展示已有轨迹'],
                },
                'meta': {'request_id': 'rq_track'},
            },
        )

        self.assertEqual(
            [{'label': '状态'}, {'label': '状态'}],
            result['structured_content']['headers'],
        )
        self.assertEqual([['已发货', '']], result['structured_content']['rows'])
        self.assertEqual(['仅展示已有轨迹'], result['structured_content']['tips'])

    def test_empty_table_keeps_safe_headers_and_tips(self):
        result = self.presenter.present(
            'query_track',
            {
                'code': 'MCP_0000',
                'data': {
                    'columns': [{'key': 'status', 'name': '轨迹节点'}],
                    'records': [],
                    'tips': ['未查询到轨迹信息'],
                },
                'meta': {'page': 1, 'limit': 5, 'total': 0},
            },
        )

        self.assertEqual([{'label': '轨迹节点'}], result['structured_content']['headers'])
        self.assertEqual([], result['structured_content']['rows'])
        self.assertIn('未查询到轨迹信息', result['text'])

        without_tips = self.presenter.present(
            'query_track',
            {
                'code': 'MCP_0000',
                'data': {
                    'columns': [{'key': 'status', 'name': '轨迹节点'}],
                    'records': [],
                    'tips': 'not-a-list',
                },
            },
        )
        self.assertNotIn('tips', without_tips['structured_content'])

    def test_filter_options_preserve_values_for_follow_up_calls(self):
        result = self.presenter.present(
            'list_order_filter_options',
            {
                'code': 'MCP_0000',
                'data': {
                    'records': [
                        {'value': 101, 'label': '客户甲', 'code': 'CUS-A'},
                        {'value': -1, 'label': '客户仓', 'code': '-1'},
                    ],
                },
                'meta': {'page': 2, 'limit': 20, 'has_more': True},
            },
        )

        self.assertEqual(
            [[101, '客户甲', 'CUS-A'], [-1, '客户仓', '-1']],
            result['structured_content']['rows'],
        )
        self.assertEqual(
            ['可传值', '显示名称', '业务编码'],
            [item['label'] for item in result['structured_content']['headers']],
        )
        self.assertNotIn('"value"', json.dumps(result, ensure_ascii=False))

    def test_pending_filter_options_use_same_safe_shape(self):
        result = self.presenter.present(
            'list_pending_outbound_export_filter_options',
            {
                'code': 'MCP_0000',
                'data': {'records': [{'value': 'US', 'label': '美国', 'code': 'US'}]},
                'meta': {},
            },
        )

        self.assertEqual([['US', '美国', 'US']], result['structured_content']['rows'])

    def test_export_result_preserves_url_without_file_url_key(self):
        for tool_name in (
            'export_pending_outbound_orders',
            'export_out_of_province_port_data',
        ):
            with self.subTest(tool_name=tool_name):
                result = self.presenter.present(
                    tool_name,
                    {
                        'code': 'MCP_0000',
                        'data': {'file_url': 'https://files.test/result.xlsx'},
                        'meta': {'request_id': 'rq_export'},
                    },
                )

                self.assertEqual(
                    [{
                        'label': '导出文件',
                        'url': 'https://files.test/result.xlsx',
                    }],
                    result['structured_content']['files'],
                )
                self.assertNotIn('file_url', json.dumps(result, ensure_ascii=False))

    def test_known_parameter_error_uses_business_label(self):
        result = self.presenter.present(
            'query_track',
            {
                'code': 'MCP_1401',
                'msg': 'tracking_number is required',
                'data': {'tracking_number': 'bad'},
                'meta': {'request_id': 'rq_error'},
            },
        )

        self.assertTrue(result['is_error'])
        self.assertEqual('快递单号参数不正确', result['structured_content']['message'])
        self.assertNotIn('tracking_number', json.dumps(result, ensure_ascii=False))
        self.assertEqual({'request_id': 'rq_error'}, result['meta'])

    def test_system_error_hides_backend_message_and_data(self):
        result = self.presenter.present(
            'query_order_exact',
            {
                'code': 'MCP_9001',
                'msg': 'SQLSTATE table st_order order_number failed',
                'data': {'sql': 'select * from st_order'},
            },
        )

        serialized = json.dumps(result, ensure_ascii=False)
        self.assertIn('工具调用失败，请稍后重试', serialized)
        self.assertNotIn('SQLSTATE', serialized)
        self.assertNotIn('st_order', serialized)
        self.assertNotIn('order_number', serialized)

    def test_unknown_tool_and_malformed_success_fail_closed(self):
        unknown = self.presenter.present(
            'future_tool',
            {'code': 'MCP_0000', 'data': {'secret_field': 'secret'}},
        )
        malformed = self.presenter.present(
            'query_track',
            {'code': 'MCP_0000', 'data': {'records': []}},
        )

        for result in (unknown, malformed):
            self.assertTrue(result['is_error'])
            self.assertEqual('工具返回格式异常', result['structured_content']['message'])
            self.assertNotIn('secret', json.dumps(result, ensure_ascii=False))

    def test_present_exception_maps_value_error_and_hides_unknown_exception(self):
        parameter = self.presenter.present_exception(
            'query_track',
            ValueError('order_number is required'),
        )
        system = self.presenter.present_exception(
            'query_track',
            RuntimeError('SQL password leaked'),
        )

        self.assertEqual('订单号参数不正确', parameter['structured_content']['message'])
        self.assertEqual('工具调用失败，请稍后重试', system['structured_content']['message'])
        self.assertNotIn('password', json.dumps(system, ensure_ascii=False))

    def test_present_exception_handles_device_disabled_and_unknown_tool(self):
        device = self.presenter.present_exception(
            'query_track',
            RuntimeError('这台设备的 Workbuddy 配置已失效，请重新生成配置。'),
        )
        disabled = self.presenter.present_exception(
            'query_track',
            RuntimeError('tool disabled: query_track'),
        )
        unknown = self.presenter.present_exception(
            'future_tool',
            RuntimeError('secret error'),
        )

        self.assertIn('Workbuddy', device['structured_content']['message'])
        self.assertEqual('MCP_1202', disabled['structured_content']['code'])
        self.assertEqual('工具返回格式异常', unknown['structured_content']['message'])

    def test_invalid_payload_types_and_generic_errors_fail_safely(self):
        cases = (
            self.presenter.present('query_track', 'invalid'),
            self.presenter.present('query_track', {'code': 'MCP_0000', 'data': []}),
            self.presenter.present('query_track', {'code': '', 'msg': 'secret'}),
            self.presenter.present(
                'query_track',
                {'code': 'MCP_7777', 'msg': 'secret backend error'},
            ),
            self.presenter.present(
                'query_track',
                {'code': 'MCP_1301', 'msg': 'secret permission detail'},
            ),
        )

        for result in cases:
            self.assertTrue(result['is_error'])
            self.assertNotIn('secret', json.dumps(result, ensure_ascii=False))

    def test_malformed_table_parts_fail_closed(self):
        base = {
            'code': 'MCP_0000',
            'data': {
                'columns': [{'key': 'status', 'name': '状态'}],
                'records': [],
            },
        }
        bad_data = (
            {'columns': [], 'records': []},
            {'columns': [{'key': 'status', 'name': '状态'}], 'records': None},
            {'columns': ['status'], 'records': []},
            {'columns': [{'key': None, 'name': '状态'}], 'records': []},
            {'columns': [{'key': ' ', 'name': '状态'}], 'records': []},
            {'columns': [{'key': 'status', 'name': None}], 'records': []},
            {'columns': [{'key': 'status', 'name': ' '}], 'records': []},
            {'columns': [{'key': 'status', 'name': '状态'}], 'records': ['bad']},
        )

        for data in bad_data:
            with self.subTest(data=data):
                payload = dict(base)
                payload['data'] = data
                self.assertTrue(self.presenter.present('query_track', payload)['is_error'])

    def test_malformed_options_and_exports_fail_closed(self):
        cases = (
            self.presenter.present(
                'list_order_filter_options',
                {'code': 'MCP_0000', 'data': {'records': None}},
            ),
            self.presenter.present(
                'list_order_filter_options',
                {'code': 'MCP_0000', 'data': {'records': ['bad']}},
            ),
            self.presenter.present(
                'export_pending_outbound_orders',
                {'code': 'MCP_0000', 'data': {'file_url': None}},
            ),
            self.presenter.present(
                'export_pending_outbound_orders',
                {'code': 'MCP_0000', 'data': {'file_url': '  '}},
            ),
        )

        self.assertTrue(all(result['is_error'] for result in cases))

    def test_optional_text_meta_and_nested_values_cover_safe_boundaries(self):
        result = self.presenter.present(
            'query_track',
            {
                'code': 'MCP_0000',
                'data': {
                    'summary': None,
                    'columns': [
                        {'key': 'status', 'name': '状态', 'description': ''},
                        {'key': 'detail', 'name': '详情', 'description': 123},
                    ],
                    'records': [{'status': {'name': '已发货'}, 'detail': ['A']}],
                    'tips': ['', None, '有效提示'],
                },
                'meta': {'request_id': ' ', 'page': 1},
            },
        )

        self.assertFalse(result['is_error'])
        self.assertEqual({}, result['meta'])
        self.assertIn('{"name": "已发货"}', result['text'])
        self.assertIn('["A"]', result['text'])
        self.assertEqual(['有效提示'], result['structured_content']['tips'])

        no_meta = self.presenter.present(
            'list_order_filter_options',
            {'code': 'MCP_0000', 'data': {'records': []}, 'meta': None},
        )
        self.assertNotIn('pagination', no_meta['structured_content'])

    def test_parameter_error_without_known_field_is_generic(self):
        result = self.presenter.present(
            'query_track',
            {'code': 'MCP_1401', 'msg': 'invalid request'},
        )

        self.assertEqual(
            '工具参数不正确，请检查后重试',
            result['structured_content']['message'],
        )

    def test_query_order_is_explicitly_not_presented(self):
        self.assertFalse(self.presenter.handles('query_order'))
        self.assertTrue(self.presenter.handles('query_order_exact'))

    def test_unhashable_tool_names_fail_closed(self):
        for tool_name in ([], {}):
            with self.subTest(tool_name=tool_name):
                self.assertFalse(self.presenter.handles(tool_name))
                result = self.presenter.present_exception(
                    tool_name,
                    RuntimeError('secret backend error'),
                )
                self.assertTrue(result['is_error'])
                self.assertEqual(
                    '工具返回格式异常',
                    result['structured_content']['message'],
                )

    def test_safe_tool_descriptions_require_business_labels_only(self):
        safe_tools = (
            QueryOrderExactTool(),
            QueryTrackTool(),
            ListOrderFilterOptionsTool(),
            ListPendingOutboundExportFilterOptionsTool(),
            ExportPendingOutboundOrdersTool(),
            ExportOutOfProvincePortDataTool(),
        )

        for tool in safe_tools:
            with self.subTest(tool=tool.name):
                self.assertIn('不得展示内部参数名', tool.metadata()['description'])
        self.assertNotIn('不得展示内部参数名', QueryOrderTool().metadata()['description'])

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

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

        order_result = self.presenter.present(
            'query_order_exact',
            {
                'code': 'MCP_0000',
                'data': {
                    'columns': [{'key': 'order_number', 'name': '订单号'}],
                    'records': [{'order_number': 'SO-FOLLOW-UP'}],
                },
            },
        )
        option_result = self.presenter.present(
            'list_order_filter_options',
            {
                'code': 'MCP_0000',
                'data': {
                    'records': [{'value': 901, 'label': '客户甲', 'code': 'C901'}],
                },
            },
        )

        client = RecordingClient()
        QueryTrackTool(client).call(
            order_number=order_result['structured_content']['rows'][0][0],
        )
        QueryOrderExactTool(client).call(
            customer_ids=[option_result['structured_content']['rows'][0][0]],
        )

        self.assertEqual(
            ('query_track', {'page': 1, 'limit': 5, 'order_number': 'SO-FOLLOW-UP'}),
            client.calls[0],
        )
        self.assertEqual([901], client.calls[1][1]['customer_ids'])


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