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_tools_reject_unknown_backend_columns(self):
        for tool_name in (
            'query_order_exact',
            'query_track',
            'query_customs_declaration_files',
        ):
            with self.subTest(tool_name=tool_name):
                result = self.presenter.present(
                    tool_name,
                    {
                        'code': 'MCP_0000',
                        'data': {
                            'columns': [
                                {'key': 'order_id', 'name': '内部订单ID'},
                            ],
                            'records': [{'order_id': 99}],
                        },
                    },
                )

                self.assertTrue(result['is_error'])
                self.assertEqual(
                    '工具返回格式异常',
                    result['structured_content']['message'],
                )
                self.assertNotIn('99', result['text'])

    def test_table_tools_use_fixed_labels_instead_of_backend_labels(self):
        result = self.presenter.present(
            'query_customs_declaration_files',
            {
                'code': 'MCP_0000',
                'data': {
                    'columns': [
                        {'key': 'order_number', 'name': '内部名称'},
                    ],
                    'records': [{'order_number': 'ORD001'}],
                },
            },
        )

        self.assertFalse(result['is_error'])
        self.assertEqual(
            [{'label': '订单号'}],
            result['structured_content']['headers'],
        )
        self.assertNotIn('内部名称', result['text'])

    def test_table_tool_without_column_definition_fails_closed(self):
        definition = self.presenter.TABLE_COLUMNS.pop('query_track')
        try:
            result = self.presenter.present(
                'query_track',
                {
                    'code': 'MCP_0000',
                    'data': {
                        'columns': [{'key': 'status', 'name': '轨迹节点'}],
                        'records': [],
                    },
                },
            )
        finally:
            self.presenter.TABLE_COLUMNS['query_track'] = definition

        self.assertTrue(result['is_error'])
        self.assertEqual(
            '工具返回格式异常',
            result['structured_content']['message'],
        )

    def test_table_missing_value_is_empty_and_fixed_labels_are_used(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': '轨迹节点',
                    'description': '轨迹状态，如：开始集港、离港放行、清关、配送等',
                },
                {'label': '轨迹内容', 'description': '详细描述'},
            ],
            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': '轨迹节点',
                'description': '轨迹状态，如：开始集港、离港放行、清关、配送等',
            }],
            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_port_export_so_number_error_uses_business_label(self):
        result = self.presenter.present(
            'export_out_of_province_port_data',
            {
                'code': 'MCP_1401',
                'msg': 'so_numbers must be an array',
                'data': {'so_numbers': 'bad'},
                'meta': {'request_id': 'rq_so_error'},
            },
        )

        self.assertTrue(result['is_error'])
        self.assertEqual('SO号参数不正确', result['structured_content']['message'])
        self.assertNotIn('so_numbers', json.dumps(result, ensure_ascii=False))

    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_new_business_error_codes_have_stable_messages_and_are_not_retryable(self):
        cases = {
            'MCP_1103': '员工账号不可用，请联系管理员',
            'MCP_1104': '身份信息已失效，请重新登录或重新生成设备配置',
            'MCP_1501': '目标数据不可用',
            'MCP_1502': '结果过多，请缩小查询范围',
            'MCP_1601': '没有可导出的数据',
        }

        for code, message in cases.items():
            with self.subTest(code=code):
                result = self.presenter.present(
                    'query_order_exact',
                    {'code': code, 'msg': 'internal detail', 'data': []},
                )
                self.assertEqual(code, result['structured_content']['code'])
                self.assertEqual(message, result['structured_content']['message'])
                self.assertFalse(result['structured_content']['retryable'])

    def test_unknown_business_error_code_is_normalized_to_system_error(self):
        with self.assertLogs('services.output_presenter', level='WARNING') as logs:
            result = self.presenter.present(
                'query_order_exact',
                {
                    'code': 'MCP_7777',
                    'msg': 'secret backend detail',
                    'data': [],
                    'meta': {'request_id': 'rq_unknown_code'},
                },
            )

        self.assertEqual('MCP_9001', result['structured_content']['code'])
        self.assertEqual('工具调用失败，请稍后重试', result['structured_content']['message'])
        self.assertTrue(result['structured_content']['retryable'])
        self.assertNotIn('secret', json.dumps(result, ensure_ascii=False))
        record = logs.records[0]
        self.assertEqual('rq_unknown_code', record.request_id)
        self.assertEqual('query_order_exact', record.tool_code)
        self.assertEqual('MCP_7777', record.backend_code)
        self.assertEqual('MCP_9001', record.response_code)
        self.assertEqual('UNEXPECTED_EXCEPTION', record.diagnostic_reason)

    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': 'content', 'name': '详情', 'description': 123},
                    ],
                    'records': [{'status': {'name': '已发货'}, 'content': ['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()