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()