|
|
@@ -0,0 +1,457 @@
|
|
|
+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()
|