import importlib import io import unittest from app import GatewayApp from public_gateway import PublicGatewayApp from services.output_presenter import OutputPresenter COLUMNS = [ ('order_number', '订单号'), ('customer_name', '客户'), ('product_name', '物流产品'), ('abnormal_status', '问题件状态'), ('abnormal_type', '问题件类型'), ('abnormal_desc', '问题件描述'), ('outbound_status', '排舱单状态'), ('inbound_date', '入库时间'), ('marker_name', '标记人'), ] STATUSES = [ '待处理', '放行待审核', '包裹放行', '退件待审核', '退件驳回', '已退件', ] class RecordingApiClient: def __init__(self): self.calls = [] def list_enabled_tools(self, request_id=''): return { 'code': 'MCP_0000', 'data': { 'tool_codes': [ 'query_order_abnormal_list', 'list_order_abnormal_filter_options', ], }, } 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 OrderAbnormalToolContractTest(unittest.TestCase): def tool_class(self, module_name, class_name): return getattr(importlib.import_module('tools.' + module_name), class_name) def test_query_schema_requires_status_and_forbids_identity(self): cls = self.tool_class( 'query_order_abnormal_list', 'QueryOrderAbnormalListTool' ) metadata = cls().metadata() schema = metadata['input_schema'] self.assertEqual('query_order_abnormal_list', metadata['name']) self.assertEqual('/mcp/tools/queryOrderAbnormalList', cls.route_path) self.assertFalse(schema['additionalProperties']) self.assertEqual(['abnormal_status'], schema['required']) self.assertEqual(STATUSES, schema['properties']['abnormal_status']['enum']) self.assertNotIn('default', schema['properties']['abnormal_status']) for forbidden in ( 'company_id', 'admin_id', 'is_super', 'customer_id', 'product_id', ): self.assertNotIn(forbidden, schema['properties']) self.assertIn('使用场景：', metadata['description']) self.assertIn('禁止使用：', metadata['description']) self.assertIn('号码类型不明确时必须先询问用户', metadata['description']) self.assertIn('不得根据号码格式猜测', metadata['description']) self.assertIn('不得跨字段或跨工具试查', metadata['description']) self.assertIn('admin/Order/abnormal', metadata['description']) self.assertIn('订单号可选', metadata['description']) def test_query_forwards_optional_order_numbers(self): cls = self.tool_class( 'query_order_abnormal_list', 'QueryOrderAbnormalListTool' ) client = RecordingApiClient() cls(client).call( abnormal_status='包裹放行', order_numbers=[' FMS-1 ', 'FMS-1'], page=2, limit=30, request_id='rq_abn', ) self.assertEqual( ( 'query_order_abnormal_list', '/mcp/tools/queryOrderAbnormalList', { 'abnormal_status': '包裹放行', 'order_numbers': ['FMS-1'], 'page': 2, 'limit': 30, }, 'rq_abn', ), client.calls[-1], ) cls(client).call(abnormal_status='待处理') self.assertEqual( {'abnormal_status': '待处理', 'page': 1, 'limit': 20}, client.calls[-1][2], ) with self.assertRaises(ValueError): cls(client).call(abnormal_status='结案') with self.assertRaises(RuntimeError): cls().call(abnormal_status='待处理') def test_filter_schema_and_call(self): cls = self.tool_class( 'list_order_abnormal_filter_options', 'ListOrderAbnormalFilterOptionsTool', ) metadata = cls().metadata() schema = metadata['input_schema'] self.assertEqual('list_order_abnormal_filter_options', metadata['name']) self.assertEqual( '/mcp/tools/listOrderAbnormalFilterOptions', cls.route_path ) self.assertEqual(['filter_type'], schema['required']) self.assertEqual(['问题件状态'], schema['properties']['filter_type']['enum']) self.assertIn('query_order_abnormal_list', metadata['description']) self.assertIn('禁止猜测', metadata['description']) client = RecordingApiClient() cls(client).call('问题件状态', keyword='待', page=2, limit=10) self.assertEqual( { 'filter_type': '问题件状态', 'keyword': '待', 'page': 2, 'limit': 10, }, client.calls[-1][2], ) with self.assertRaises(ValueError): cls(client).call('问题件类型') with self.assertRaises(RuntimeError): cls().call('问题件状态') with self.assertRaises(ValueError): cls(client).call('问题件状态', keyword='x' * 101) with self.assertRaises(ValueError): cls(client).call('问题件状态', page=0) with self.assertRaises(ValueError): cls(client).call('问题件状态', limit=True) def test_query_covers_remaining_validation_branches(self): cls = self.tool_class( 'query_order_abnormal_list', 'QueryOrderAbnormalListTool' ) client = RecordingApiClient() with self.assertRaises(ValueError): cls(client).call( abnormal_status='待处理', order_numbers=['X%s' % i for i in range(201)], ) with self.assertRaises(ValueError): cls(client).call(abnormal_status='待处理', order_numbers='FMS-1') with self.assertRaises(ValueError): cls(client).call(abnormal_status='待处理', order_numbers=[1]) with self.assertRaises(ValueError): cls(client).call(abnormal_status='待处理', order_numbers=[' ']) with self.assertRaises(ValueError): cls(client).call( abnormal_status='待处理', order_numbers=['X' * 101], ) with self.assertRaises(ValueError): cls(client).call(abnormal_status='待处理', page=0) with self.assertRaises(ValueError): cls(client).call(abnormal_status='待处理', limit=True) with self.assertRaises(ValueError): cls(client).call(abnormal_status='待处理', order_numbers=[]) with self.assertRaises(ValueError): cls(client).call(abnormal_status='待处理', page=101) def test_cli_forwards_status_and_order_numbers(self): client = RecordingApiClient() app = GatewayApp(api_client=client) with self.assertRaises(ValueError): app.run_cli([ 'call', '--tool', 'query_order_abnormal_list', ], stdout=io.StringIO()) app.run_cli([ 'call', '--tool', 'query_order_abnormal_list', '--abnormal-status', '退件待审核', '--order-numbers', ' FMS-1,FMS-2 ', ], stdout=io.StringIO()) self.assertEqual( { 'abnormal_status': '退件待审核', 'order_numbers': ['FMS-1', 'FMS-2'], 'page': 1, 'limit': 20, }, client.calls[-1][2], ) app.run_cli([ 'call', '--tool', 'query_order_abnormal_list', '--abnormal-status', '待处理', ], stdout=io.StringIO()) self.assertEqual( {'abnormal_status': '待处理', 'page': 1, 'limit': 20}, client.calls[-1][2], ) app.run_cli([ 'call', '--tool', 'list_order_abnormal_filter_options', '--filter-type', '问题件状态', ], stdout=io.StringIO()) self.assertEqual('list_order_abnormal_filter_options', client.calls[-1][0]) with self.assertRaises(ValueError): app.run_cli([ 'call', '--tool', 'list_order_abnormal_filter_options', ], stdout=io.StringIO()) def test_presenter_renders_nine_columns_and_fails_closed(self): presenter = OutputPresenter() meta = { 'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_x', } columns = [{'key': key, 'name': name} for key, name in COLUMNS] record = {key: 'v' for key, _ in COLUMNS} record['abnormal_status'] = '待处理' ok = presenter.present('query_order_abnormal_list', { 'code': 'MCP_0000', 'data': { 'abnormal_status': '待处理', 'columns': columns, 'records': [record], }, 'meta': meta, }) self.assertFalse(ok['is_error']) self.assertEqual( ['订单号', '客户', '物流产品', '问题件状态', '问题件类型', '问题件描述', '排舱单状态', '入库时间', '标记人'], [item['label'] for item in ok['structured_content']['headers']], ) self.assertTrue(presenter.present('query_order_abnormal_list', { 'code': 'MCP_0000', 'data': { 'abnormal_status': '待处理', 'columns': [], 'records': [], 'extra': 1, }, 'meta': meta, })['is_error']) self.assertTrue(presenter.present('query_order_abnormal_list', { 'code': 'MCP_0000', 'data': { 'abnormal_status': '结案', 'columns': columns, 'records': [], }, 'meta': meta, })['is_error']) bad_columns = list(columns) bad_columns[0] = {'key': 'order_number', 'name': '错'} self.assertTrue(presenter.present('query_order_abnormal_list', { 'code': 'MCP_0000', 'data': { 'abnormal_status': '待处理', 'columns': bad_columns, 'records': [], }, 'meta': meta, })['is_error']) bad_record = dict(record) bad_record['order_number'] = 1 self.assertTrue(presenter.present('query_order_abnormal_list', { 'code': 'MCP_0000', 'data': { 'abnormal_status': '待处理', 'columns': columns, 'records': [bad_record], }, 'meta': meta, })['is_error']) self.assertTrue(presenter.present('query_order_abnormal_list', { 'code': 'MCP_0000', 'data': { 'abnormal_status': '待处理', 'columns': columns, 'records': 'bad', }, 'meta': meta, })['is_error']) extra_record = dict(record) extra_record['deal_remark'] = 'x' self.assertTrue(presenter.present('query_order_abnormal_list', { 'code': 'MCP_0000', 'data': { 'abnormal_status': '待处理', 'columns': columns, 'records': [extra_record], }, 'meta': meta, })['is_error']) self.assertTrue(presenter.present('query_order_abnormal_list', { 'code': 'MCP_0000', 'data': { 'abnormal_status': '待处理', 'columns': None, 'records': [], }, 'meta': meta, })['is_error']) self.assertTrue(presenter.present('query_order_abnormal_list', { 'code': 'MCP_0000', 'data': { 'abnormal_status': '待处理', 'columns': columns, 'records': [None], }, 'meta': meta, })['is_error']) self.assertTrue(presenter.present('query_order_abnormal_list', { 'code': 'MCP_0000', 'data': { 'abnormal_status': '待处理', 'columns': columns, 'records': [], }, 'meta': { 'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_x', 'total': 1, }, })['is_error']) options = presenter.present('list_order_abnormal_filter_options', { 'code': 'MCP_0000', 'data': { 'records': [ {'value': '待处理', 'label': '待处理', 'code': ''}, ], }, 'meta': meta, }) self.assertFalse(options['is_error']) def test_local_public_registry_includes_both_tools(self): local = GatewayApp().registered_tool_names() public = PublicGatewayApp(None, None).registered_tool_names() self.assertEqual(local, public) self.assertEqual(30, len(local)) self.assertEqual(29, len(OutputPresenter.SAFE_TOOLS)) self.assertIn('query_order_abnormal_list', local) self.assertIn('list_order_abnormal_filter_options', local)