import importlib import io import json import os import unittest from app import GatewayApp from public_gateway import PublicGatewayApp from services.output_presenter import OutputPresenter class RecordingApiClient: def __init__(self, enabled=True): self.enabled = enabled self.last_call = None def list_enabled_tools(self): return { 'code': 'MCP_0000', 'data': {'tool_codes': ( ['query_customs_declaration_files'] if self.enabled else [] )}, } def call_tool(self, tool_code, route_path, payload, request_id): self.last_call = { 'tool_code': tool_code, 'route_path': route_path, 'payload': payload, 'request_id': request_id, } return {'code': 'MCP_0000', 'data': {'columns': [], 'records': []}} class PublicSessionStore: def get(self, gateway_session_id): if gateway_session_id == 'GWS_test': return {'mcp_token': 'MT_test', 'company_id': 7, 'admin_id': 9} return None class PublicApiClient: def __init__(self, enabled=True): self.enabled = enabled self.calls = [] def list_enabled_tools(self, token): return { 'code': 'MCP_0000', 'data': {'tool_codes': ( ['query_customs_declaration_files'] if self.enabled else [] )}, } def call_tool(self, **kwargs): self.calls.append(kwargs) return {'code': 'MCP_0000'} class QueryCustomsDeclarationFilesToolTest(unittest.TestCase): def tool_class(self): path = os.path.join( os.path.dirname(os.path.dirname(__file__)), 'tools', 'query_customs_declaration_files.py', ) self.assertTrue(os.path.exists(path), path) module = importlib.import_module( 'tools.query_customs_declaration_files' ) return module.QueryCustomsDeclarationFilesTool def test_metadata_teaches_ai_exact_order_number_semantics(self): metadata = self.tool_class()().metadata() schema = metadata['input_schema'] properties = schema['properties'] self.assertFalse(schema['additionalProperties']) self.assertEqual([ {'required': ['outbound_numbers']}, {'required': ['order_numbers']}, ], schema['oneOf']) for field in ('outbound_numbers', 'order_numbers'): prop = properties[field] self.assertEqual('array', prop['type']) self.assertEqual('string', prop['items']['type']) self.assertEqual(1, prop['minItems']) self.assertEqual(100, prop['maxItems']) self.assertNotIn('uniqueItems', prop) order_description = properties['order_numbers']['description'] for phrase in ( '订单号', '后台订单列表', '排舱详情', '不是系统单号', 'order_id/id', '不是客户参考号', '快递单号', '排舱单号', ): self.assertIn(phrase, order_description) self.assertNotIn('系统订单号', order_description) description = metadata['description'] for phrase in ( '用户明确说“订单号”', '用户明确说“排舱单号”', '没有明确说明是订单号还是排舱单号', '必须先提问，让用户选择“订单号”或“排舱单号”', '只说“单号”时必须先追问', '不是系统单号', '用户说“系统单号”时也不得当作订单号', '不得根据号码格式猜测', '不得跨字段重试', '不得展示内部参数名', ): self.assertIn(phrase, description) self.assertNotIn('系统订单号', description) def test_outbound_batch_normalizes_and_forwards_supported_fields(self): client = RecordingApiClient() tool = self.tool_class()(api_client=client) result = tool.call( outbound_numbers=[' PC001 ', 'PC002', 'PC001'], page=2, limit=50, request_id='rq_customs', ) self.assertEqual('MCP_0000', result['code']) self.assertEqual( 'query_customs_declaration_files', client.last_call['tool_code'], ) self.assertEqual( '/mcp/tools/queryCustomsDeclarationFiles', client.last_call['route_path'], ) self.assertEqual({ 'outbound_numbers': ['PC001', 'PC002'], 'page': 2, 'limit': 50, }, client.last_call['payload']) def test_order_batch_uses_order_numbers_business_field(self): client = RecordingApiClient() tool = self.tool_class()(api_client=client) tool.call(order_numbers=[' ORD001 ', 'ORD002']) self.assertEqual({ 'order_numbers': ['ORD001', 'ORD002'], 'page': 1, 'limit': 20, }, client.last_call['payload']) def test_call_rejects_mixed_missing_and_invalid_batches(self): tool = self.tool_class()(api_client=RecordingApiClient()) invalid = ( {}, {'outbound_numbers': ['PC001'], 'order_numbers': ['ORD001']}, {'order_numbers': []}, {'order_numbers': 'ORD001'}, {'order_numbers': ['']}, {'order_numbers': ['ORD001', 2]}, {'order_numbers': ['X' * 101]}, {'order_numbers': ['ORD{0}'.format(i) for i in range(101)]}, {'order_numbers': ['ORD001'], 'page': 0}, {'order_numbers': ['ORD001'], 'page': 101}, {'order_numbers': ['ORD001'], 'limit': 101}, {'order_numbers': ['ORD001'], 'page': True}, {'order_numbers': ['ORD001'], 'page': 'not-a-number'}, ) for arguments in invalid: with self.subTest(arguments=arguments): with self.assertRaises(ValueError): tool.call(**arguments) with self.assertRaisesRegex(RuntimeError, 'api client is required'): self.tool_class()().call(order_numbers=['ORD001']) def test_local_and_public_gateways_follow_dynamic_registry(self): local_enabled = { item['name'] for item in GatewayApp(api_client=RecordingApiClient()).list_tools() } local_disabled = { item['name'] for item in GatewayApp( api_client=RecordingApiClient(enabled=False) ).list_tools() } public_enabled = { item['name'] for item in PublicGatewayApp( PublicSessionStore(), PublicApiClient(), ).list_tools('GWS_test') } self.assertIn('query_customs_declaration_files', local_enabled) self.assertNotIn('query_customs_declaration_files', local_disabled) self.assertIn('query_customs_declaration_files', public_enabled) def test_public_gateway_forwards_request_scoped_token_without_company_input(self): api_client = PublicApiClient() gateway = PublicGatewayApp(PublicSessionStore(), api_client) self.assertIn( 'query_customs_declaration_files', gateway.registered_tool_names(), ) gateway.call_tool( 'GWS_test', 'query_customs_declaration_files', {'order_numbers': ['ORD001']}, request_id='rq_public_customs', ) call = api_client.calls[0] self.assertEqual('MT_test', call['token']) self.assertEqual( '/mcp/tools/queryCustomsDeclarationFiles', call['route_path'], ) self.assertEqual({'order_numbers': ['ORD001']}, call['payload']) self.assertNotIn('company_id', call['payload']) def test_output_presenter_hides_internal_fields_and_keeps_links(self): result = OutputPresenter().present( 'query_customs_declaration_files', { 'code': 'MCP_0000', 'data': { 'summary': '当前页返回 1 个订单的报关资料', 'columns': [ {'key': 'outbound_number', 'name': '排舱单号'}, {'key': 'order_number', 'name': '订单号'}, {'key': 'file_name', 'name': '文件名'}, {'key': 'file_url', 'name': '文件链接'}, ], 'records': [{ 'outbound_number': 'PC001', 'order_number': 'ORD001', 'file_name': '报关单.pdf', 'file_url': 'https://files.test/a.pdf', 'order_id': 99, }], }, 'meta': { 'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_present', }, }, ) self.assertFalse(result['is_error']) self.assertEqual( ['排舱单号', '订单号', '文件名', '文件链接'], [header['label'] for header in result['structured_content']['headers']], ) self.assertIn('https://files.test/a.pdf', result['text']) serialized = json.dumps(result, ensure_ascii=False) for internal in ( 'outbound_number', 'order_number', 'file_name', 'file_url', 'order_id', ): self.assertNotIn(internal, serialized) def test_cli_forwards_batch_arguments(self): client = RecordingApiClient() stdout = io.StringIO() app = GatewayApp(api_client=client) self.assertIn( 'query_customs_declaration_files', app.registered_tool_names(), ) result = app.run_cli([ 'call', '--tool', 'query_customs_declaration_files', '--order-numbers', 'ORD001,ORD002', '--page', '2', '--limit', '10', ], stdout=stdout) self.assertEqual(0, result) self.assertEqual({ 'order_numbers': ['ORD001', 'ORD002'], 'page': 2, 'limit': 10, }, client.last_call['payload']) app.run_cli([ 'call', '--tool', 'query_customs_declaration_files', '--outbound-numbers', 'PC001,PC002', ], stdout=io.StringIO()) self.assertEqual({ 'outbound_numbers': ['PC001', 'PC002'], 'page': 1, 'limit': 20, }, client.last_call['payload']) if __name__ == '__main__': unittest.main()