mcp/tests/test_order_detail_tool.py

import json
import unittest

from app import GatewayApp
from public_gateway import PublicGatewayApp
from services.output_presenter import OutputPresenter
from tools.query_order_detail import QueryOrderDetailTool


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

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


class OrderDetailToolTest(unittest.TestCase):
    def test_schema_uses_only_chinese_sections_and_rejects_extra_fields(self):
        schema = QueryOrderDetailTool().metadata()['input_schema']
        self.assertEqual(['order_number'], schema['required'])
        self.assertFalse(schema['additionalProperties'])
        self.assertEqual([
            '订单概览', '箱单信息', '箱单商品', 'DW授权信息', '附件信息',
            '入库信息', '查验信息', '订单轨迹', '操作日志',
            '应收与结算日志', '派送信息', '全部',
        ], schema['properties']['section']['enum'])
        self.assertNotIn('order_id', schema['properties'])
        self.assertNotIn('company_id', schema['properties'])
        self.assertEqual('全部', schema['properties']['section']['default'])

    def test_call_forwards_normalized_payload_to_exact_route(self):
        client = RecordingClient()
        result = QueryOrderDetailTool(client).call(
            order_number=' ORD001 ', section='附件信息', page=2, limit=10,
            request_id='rq_detail',
        )
        self.assertEqual('MCP_0000', result['code'])
        self.assertEqual([(
            'query_order_detail', '/mcp/tools/queryOrderDetail',
            {'order_number': 'ORD001', 'section': '附件信息', 'page': 2, 'limit': 10},
            'rq_detail',
        )], client.calls)

    def test_call_rejects_missing_invalid_and_out_of_range_parameters(self):
        tool = QueryOrderDetailTool()
        with self.assertRaisesRegex(RuntimeError, 'api client'):
            tool.call(order_number='ORD001')

        tool = QueryOrderDetailTool(RecordingClient())
        invalid_calls = (
            ({'order_number': None}, 'order_number'),
            ({'order_number': ' '}, 'order_number'),
            ({'order_number': 'A' * 101}, 'order_number'),
            ({'order_number': 'ORD001', 'section': None}, 'section'),
            ({'order_number': 'ORD001', 'section': 'unknown'}, 'section'),
            ({'order_number': 'ORD001', 'page': True}, 'page'),
            ({'order_number': 'ORD001', 'page': '01'}, 'page'),
            ({'order_number': 'ORD001', 'page': 'x'}, 'page'),
            ({'order_number': 'ORD001', 'limit': 0}, 'limit'),
            ({'order_number': 'ORD001', 'limit': 101}, 'limit'),
        )
        for kwargs, expected in invalid_calls:
            with self.subTest(kwargs=kwargs):
                with self.assertRaisesRegex(ValueError, expected):
                    tool.call(**kwargs)

        tool.call(order_number='ORD001', page='2', limit='100')
        self.assertEqual(2, tool.api_client.calls[-1][2]['page'])
        self.assertEqual(100, tool.api_client.calls[-1][2]['limit'])

    def test_call_forwards_all_section_without_changing_pagination_contract(self):
        client = RecordingClient()
        QueryOrderDetailTool(client).call(
            order_number='ORD001', section='全部', page=1, limit=20,
            request_id='rq_all',
        )
        self.assertEqual({
            'order_number': 'ORD001', 'section': '全部', 'page': 1, 'limit': 20,
        }, client.calls[0][2])

    def test_call_defaults_to_all_when_section_is_omitted(self):
        client = RecordingClient()
        QueryOrderDetailTool(client).call(order_number='ORD001')
        self.assertEqual('全部', client.calls[0][2]['section'])

    def test_local_and_public_registries_include_order_detail(self):
        local = GatewayApp(api_client=object())
        public = PublicGatewayApp(session_store=object(), api_client=object())
        self.assertIn('query_order_detail', local.registered_tool_names())
        self.assertIn('query_order_detail', public.registered_tool_names())

    def test_presenter_outputs_attachment_links_with_only_chinese_keys(self):
        presenter = OutputPresenter()
        result = presenter.present('query_order_detail', {
            'code': 'MCP_0000',
            'data': {
                'section': 'attachments',
                'order_number': 'ORD001',
                'payload': {'records': [{
                    'category': '订单附件', 'file_name': 'photo.jpg',
                    'file_extension': 'jpg', 'is_image': True,
                    'preview_url': 'https://files.example/photo.jpg',
                    'download_url': 'https://files.example/download/photo.jpg',
                    'cost_name': '',
                }]},
            },
            'meta': {'request_id': 'rq_detail', 'page': 1, 'limit': 20, 'has_more': False},
        })
        self.assertFalse(result['is_error'])
        content = result['structured_content']
        self.assertEqual('ORD001', content['订单号'])
        self.assertEqual('附件信息', content['详情模块'])
        self.assertEqual('photo.jpg', content['明细'][0]['文件名称'])
        self.assertEqual('https://files.example/photo.jpg', content['明细'][0]['预览链接'])
        serialized = json.dumps(content, ensure_ascii=False) + result['text']
        for internal in ('order_number', 'section', 'payload', 'file_name', 'preview_url', 'page', 'limit', 'has_more'):
            self.assertNotIn(internal, serialized)

    def test_presenter_outputs_complete_overview_and_translates_enums(self):
        presenter = OutputPresenter()
        result = presenter.present('query_order_detail', {
            'code': 'MCP_0000',
            'data': {
                'section': 'overview', 'order_number': 'ORD001',
                'payload': {
                    'status_nodes': [{
                        'stage': '起运段', 'label': '已提交', 'state': 'completed',
                        'occurred_at': '2026-01-01', 'time_kind': 'actual',
                    }],
                    'order_info': self.fixed_values(presenter.ORDER_DETAIL_FIELDS['overview']['order_info']),
                    'freight_info': self.fixed_values(presenter.ORDER_DETAIL_FIELDS['overview']['freight_info']),
                    'package_summary': self.fixed_values(presenter.ORDER_DETAIL_FIELDS['overview']['package_summary']),
                    'trader_info': self.fixed_values(presenter.ORDER_DETAIL_FIELDS['overview']['trader_info']),
                },
            },
            'meta': {'request_id': 'rq_overview'},
        })
        self.assertFalse(result['is_error'])
        content = result['structured_content']
        self.assertEqual('已完成', content['状态节点'][0]['状态'])
        self.assertEqual('实际', content['状态节点'][0]['时间类型'])
        self.assertEqual(
            ['订单号', '详情模块', '状态节点', '订单信息', '货运信息', '箱单汇总', '进出口商'],
            list(content.keys()),
        )

    def test_presenter_outputs_all_sections_as_strict_grouped_content(self):
        presenter = OutputPresenter()
        payload = self.valid_all_payload(presenter)

        result = presenter.present('query_order_detail', {
            'code': 'MCP_0000',
            'data': {'section': 'all', 'order_number': 'ORD001', 'payload': payload},
            'meta': {'request_id': 'rq_all'},
        })
        self.assertFalse(result['is_error'])
        content = result['structured_content']
        self.assertEqual('全部', content['详情模块'])
        self.assertIn('订单概览', content)
        self.assertIn('箱单信息', content)
        self.assertIn('分页', content['箱单信息'])
        serialized = json.dumps(content, ensure_ascii=False) + result['text']
        for internal in ('order_number', 'section', 'payload', 'preview_url', 'has_more'):
            self.assertNotIn(internal, serialized)

    def valid_all_payload(self, presenter):
        overview = {
            'status_nodes': [{
                'stage': '起运段', 'label': '已提交', 'state': 'completed',
                'occurred_at': '2026-01-01', 'time_kind': 'actual',
            }],
            'order_info': self.fixed_values(presenter.ORDER_DETAIL_FIELDS['overview']['order_info']),
            'freight_info': self.fixed_values(presenter.ORDER_DETAIL_FIELDS['overview']['freight_info']),
            'package_summary': self.fixed_values(presenter.ORDER_DETAIL_FIELDS['overview']['package_summary']),
            'trader_info': self.fixed_values(presenter.ORDER_DETAIL_FIELDS['overview']['trader_info']),
        }
        pagination = {'page': 1, 'limit': 20, 'has_more': False}
        payload = {'overview': overview}
        for section in presenter.ORDER_DETAIL_SECTIONS:
            if section in ('overview', 'all'):
                continue
            if section == 'inbound':
                value = {
                    'summary': self.fixed_values(presenter.ORDER_DETAIL_FIELDS['inbound_summary']),
                    'records': [],
                }
            elif section == 'delivery':
                value = {
                    'summary': self.fixed_values(presenter.ORDER_DETAIL_FIELDS['delivery_summary']),
                    'records': [],
                }
            else:
                value = {'records': []}
            value['pagination'] = dict(pagination)
            payload[section] = value
        return payload

    def test_presenter_rejects_malformed_all_section_groups(self):
        presenter = OutputPresenter()
        cases = []

        missing = self.valid_all_payload(presenter)
        del missing['checks']
        cases.append(missing)

        bad_overview = self.valid_all_payload(presenter)
        bad_overview['overview'] = []
        cases.append(bad_overview)

        bad_node = self.valid_all_payload(presenter)
        bad_node['overview']['status_nodes'][0]['state'] = 'unknown'
        cases.append(bad_node)

        bad_node_fields = self.valid_all_payload(presenter)
        del bad_node_fields['overview']['status_nodes'][0]['stage']
        cases.append(bad_node_fields)

        bad_group = self.valid_all_payload(presenter)
        bad_group['overview']['order_info']['unexpected'] = ''
        cases.append(bad_group)

        bad_pagination = self.valid_all_payload(presenter)
        bad_pagination['packages']['pagination']['page'] = True
        cases.append(bad_pagination)

        missing_pagination = self.valid_all_payload(presenter)
        del missing_pagination['packages']['pagination']
        cases.append(missing_pagination)

        bad_inbound_keys = self.valid_all_payload(presenter)
        bad_inbound_keys['inbound']['extra'] = ''
        cases.append(bad_inbound_keys)

        bad_inbound_summary = self.valid_all_payload(presenter)
        bad_inbound_summary['inbound']['summary']['extra'] = ''
        cases.append(bad_inbound_summary)

        bad_delivery_keys = self.valid_all_payload(presenter)
        bad_delivery_keys['delivery']['extra'] = ''
        cases.append(bad_delivery_keys)

        bad_delivery_summary = self.valid_all_payload(presenter)
        bad_delivery_summary['delivery']['summary']['extra'] = ''
        cases.append(bad_delivery_summary)

        bad_rows = self.valid_all_payload(presenter)
        bad_rows['packages']['records'] = {}
        cases.append(bad_rows)

        bad_generic_keys = self.valid_all_payload(presenter)
        bad_generic_keys['packages']['extra'] = ''
        cases.append(bad_generic_keys)

        for payload in cases:
            with self.subTest(payload_keys=list(payload)):
                result = presenter.present('query_order_detail', {
                    'code': 'MCP_0000',
                    'data': {'section': 'all', 'order_number': 'ORD001', 'payload': payload},
                    'meta': {'request_id': 'rq_all_bad'},
                })
                self.assertTrue(result['is_error'])

    def test_presenter_outputs_inbound_and_delivery_sections(self):
        presenter = OutputPresenter()
        inbound_mapping = presenter.ORDER_DETAIL_FIELDS['inbound']
        inbound_rows = []
        for mode, expected_label in (
            ('single_box', '单箱入库重量'),
            ('total', '总重量KG'),
        ):
            row = self.fixed_values(inbound_mapping)
            row['weight_mode'] = mode
            row['weight'] = '12.50'
            inbound_rows.append(row)
        inbound = presenter.present('query_order_detail', {
            'code': 'MCP_0000',
            'data': {
                'section': 'inbound', 'order_number': ' ORD001 ',
                'payload': {
                    'summary': self.fixed_values(
                        presenter.ORDER_DETAIL_FIELDS['inbound_summary']
                    ),
                    'records': inbound_rows,
                },
            },
            'meta': {'page': 1, 'limit': 20, 'has_more': True},
        })
        self.assertFalse(inbound['is_error'])
        self.assertEqual('ORD001', inbound['structured_content']['订单号'])
        for index, label in enumerate(('单箱入库重量', '总重量KG')):
            self.assertEqual('12.50', inbound['structured_content']['明细'][index][label])

        delivery = presenter.present('query_order_detail', {
            'code': 'MCP_0000',
            'data': {
                'section': 'delivery', 'order_number': 'ORD001',
                'payload': {
                    'summary': self.fixed_values(
                        presenter.ORDER_DETAIL_FIELDS['delivery_summary']
                    ),
                    'records': [self.fixed_values(
                        presenter.ORDER_DETAIL_FIELDS['delivery']
                    )],
                },
            },
            'meta': {'page': 2, 'limit': 10, 'has_more': False},
        })
        self.assertFalse(delivery['is_error'])
        self.assertIn('派送汇总', delivery['structured_content'])

    def test_package_item_product_image_is_exposed_as_count_only(self):
        presenter = OutputPresenter()
        mapping = presenter.ORDER_DETAIL_FIELDS['package_items']
        self.assertEqual('商品图片数量', mapping.get('product_image_count'))
        row = self.fixed_values(mapping)
        row['product_image_count'] = 1
        result = presenter.present('query_order_detail', {
            'code': 'MCP_0000',
            'data': {
                'section': 'package_items', 'order_number': 'ORD001',
                'payload': {'records': [row]},
            },
            'meta': {'page': 1, 'limit': 20, 'has_more': False},
        })
        self.assertFalse(result['is_error'])
        self.assertEqual(1, result['structured_content']['明细'][0]['商品图片数量'])
        self.assertNotIn('image_url', json.dumps(result, ensure_ascii=False))

    def test_unknown_missing_or_extra_fields_fail_closed(self):
        presenter = OutputPresenter()
        cases = [
            {'section': 'secret', 'order_number': 'ORD001', 'payload': {'records': []}},
            {'section': 'attachments', 'order_number': 'ORD001', 'payload': {'records': [{'category': 'x'}]}},
            {'section': 'attachments', 'order_number': 'ORD001', 'payload': {'records': [{
                'category': '', 'file_name': '', 'file_extension': '', 'is_image': False,
                'preview_url': '', 'download_url': '', 'cost_name': '', 'secret_id': 9,
            }]}},
        ]
        for data in cases:
            with self.subTest(data=data):
                result = presenter.present('query_order_detail', {
                    'code': 'MCP_0000', 'data': data, 'meta': {},
                })
                self.assertTrue(result['is_error'])
                self.assertNotIn('secret_id', json.dumps(result, ensure_ascii=False))

    def test_presenter_rejects_every_invalid_order_detail_shape(self):
        presenter = OutputPresenter()
        valid_meta = {'page': 1, 'limit': 20, 'has_more': False}
        overview_groups = {
            'status_nodes': [],
            'order_info': self.fixed_values(presenter.ORDER_DETAIL_FIELDS['overview']['order_info']),
            'freight_info': self.fixed_values(presenter.ORDER_DETAIL_FIELDS['overview']['freight_info']),
            'package_summary': self.fixed_values(presenter.ORDER_DETAIL_FIELDS['overview']['package_summary']),
            'trader_info': self.fixed_values(presenter.ORDER_DETAIL_FIELDS['overview']['trader_info']),
        }
        invalid_results = [
            presenter._present_order_detail({'section': 'attachments'}, valid_meta, {}),
            presenter._present_order_detail(
                {'section': 'unknown', 'order_number': 'ORD', 'payload': {}}, valid_meta, {}
            ),
            presenter._present_order_detail(
                {'section': 'attachments', 'order_number': 1, 'payload': {}}, valid_meta, {}
            ),
            presenter._present_order_detail(
                {'section': 'attachments', 'order_number': ' ', 'payload': {}}, valid_meta, {}
            ),
            presenter._present_order_detail(
                {'section': 'attachments', 'order_number': 'ORD', 'payload': []}, valid_meta, {}
            ),
            presenter._present_order_detail(
                {'section': 'overview', 'order_number': 'ORD', 'payload': {'status_nodes': []}}, {}, {}
            ),
            presenter._present_order_detail(
                {'section': 'overview', 'order_number': 'ORD', 'payload': dict(overview_groups, status_nodes={})}, {}, {}
            ),
        ]

        bad_node_groups = dict(overview_groups)
        bad_node_groups['status_nodes'] = [{'stage': 'x'}]
        invalid_results.append(presenter._present_order_detail(
            {'section': 'overview', 'order_number': 'ORD', 'payload': bad_node_groups}, {}, {}
        ))
        bad_enum_groups = dict(overview_groups)
        bad_enum_groups['status_nodes'] = [{
            'stage': 'x', 'label': 'x', 'state': 'secret',
            'occurred_at': '', 'time_kind': 'empty',
        }]
        invalid_results.append(presenter._present_order_detail(
            {'section': 'overview', 'order_number': 'ORD', 'payload': bad_enum_groups}, {}, {}
        ))
        bad_group = dict(overview_groups)
        bad_group['order_info'] = {}
        invalid_results.append(presenter._present_order_detail(
            {'section': 'overview', 'order_number': 'ORD', 'payload': bad_group}, {}, {}
        ))

        for section, summary_key in (
            ('inbound', 'inbound_summary'),
            ('delivery', 'delivery_summary'),
        ):
            mapping = presenter.ORDER_DETAIL_FIELDS[summary_key]
            invalid_results.extend([
                presenter._present_order_detail(
                    {'section': section, 'order_number': 'ORD', 'payload': {'records': []}}, valid_meta, {}
                ),
                presenter._present_order_detail(
                    {'section': section, 'order_number': 'ORD', 'payload': {'summary': {}, 'records': []}}, valid_meta, {}
                ),
                presenter._present_order_detail(
                    {'section': section, 'order_number': 'ORD', 'payload': {
                        'summary': self.fixed_values(mapping), 'records': []
                    }}, {}, {}
                ),
            ])

        invalid_results.extend([
            presenter._present_order_detail(
                {'section': 'attachments', 'order_number': 'ORD', 'payload': {}}, valid_meta, {}
            ),
            presenter._present_order_detail(
                {'section': 'attachments', 'order_number': 'ORD', 'payload': {'records': {}}}, valid_meta, {}
            ),
            presenter._present_order_detail(
                {'section': 'attachments', 'order_number': 'ORD', 'payload': {'records': []}}, {}, {}
            ),
        ])
        self.assertTrue(all(result['is_error'] for result in invalid_results))

    def test_order_detail_translation_helpers_reject_nested_and_invalid_values(self):
        presenter = OutputPresenter()
        self.assertIsNone(presenter._translate_order_detail_rows('attachments', {}))
        self.assertIsNone(presenter._translate_order_detail_rows('unknown', []))
        self.assertIsNone(presenter._translate_order_detail_rows('attachments', [{}]))

        inbound = self.fixed_values(presenter.ORDER_DETAIL_FIELDS['inbound'])
        inbound['weight_mode'] = 'invalid'
        self.assertIsNone(presenter._translate_order_detail_rows('inbound', [inbound]))

        self.assertIsNone(presenter._translate_exact([], {'key': '标签'}))
        self.assertIsNone(presenter._translate_exact({'wrong': 1}, {'key': '标签'}))
        self.assertIsNone(presenter._translate_exact({'key': []}, {'key': '标签'}))
        self.assertEqual({'标签': ''}, presenter._translate_exact({'key': None}, {'key': '标签'}))
        self.assertEqual({}, presenter._translate_exact({'key': 'hidden'}, {'key': None}))

        invalid_meta = (
            None,
            {},
            {'page': True, 'limit': 20, 'has_more': False},
            {'page': 0, 'limit': 20, 'has_more': False},
            {'page': 1, 'limit': True, 'has_more': False},
            {'page': 1, 'limit': 0, 'has_more': False},
            {'page': 1, 'limit': 20, 'has_more': 1},
        )
        for meta in invalid_meta:
            with self.subTest(meta=meta):
                self.assertIsNone(presenter._order_detail_pagination(meta))

    def test_parameter_errors_use_chinese_business_labels(self):
        presenter = OutputPresenter()
        for backend, expected in (
            ('order_number is required', '订单号参数不正确'),
            ('section is invalid', '详情模块参数不正确'),
            ('page is invalid', '页码参数不正确'),
            ('limit is invalid', '每页数量参数不正确'),
        ):
            result = presenter.present('query_order_detail', {
                'code': 'MCP_1401', 'msg': backend, 'data': {},
            })
            self.assertEqual(expected, result['structured_content']['message'])
            self.assertNotIn(backend.split()[0], json.dumps(result, ensure_ascii=False))

    @staticmethod
    def fixed_values(mapping):
        return {key: '' for key in mapping}


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