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