import inspect import unittest from io import StringIO import app as gateway_app_module from app import GatewayApp, parse_int_list from public_gateway import PublicGatewayApp from tools.query_order_exact import QueryOrderExactTool class RecordingApiClient: def __init__(self): self.last_call = None 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'} def list_enabled_tools(self): return { 'code': 'MCP_0000', 'data': {'tool_codes': ['query_order_exact']}, } class PublicSessionStore: def get(self, gateway_session_id): if gateway_session_id == 'GWS_test': return {'mcp_token': 'MT_test'} return None class PublicApiClient: def list_enabled_tools(self, token): return { 'code': 'MCP_0000', 'data': {'tool_codes': ['query_order_exact']}, } class QueryOrderExactToolTest(unittest.TestCase): def test_local_and_public_gateways_register_tool(self): local_names = { tool['name'] for tool in GatewayApp(api_client=RecordingApiClient()).list_tools() } public_names = { tool['name'] for tool in PublicGatewayApp( PublicSessionStore(), PublicApiClient(), ).list_tools('GWS_test') } self.assertIn('query_order_exact', local_names) self.assertIn('query_order_exact', public_names) def test_parse_int_list_accepts_comma_separated_ids(self): self.assertEqual([9, 12], parse_int_list('9, 12')) self.assertEqual([], parse_int_list('')) def test_metadata_exposes_all_exact_filters(self): schema = QueryOrderExactTool().metadata()['input_schema'] for field in ( 'order_number', 'reference_number', 'tracking_number', 'outbound_number', 'container_code', 'so_number', 'shipment_id', 'receiver_country', 'product_ids', 'customer_ids', 'sales_id', 'warehouse_ids', 'department_id', 'inbound_date_start', 'inbound_date_end', 'outbound_date_start', 'outbound_date_end', 'page', 'limit', ): self.assertIn(field, schema['properties']) for field in ( 'order_numbers', 'reference_numbers', 'tracking_numbers', 'outbound_numbers', 'container_codes', 'so_numbers', ): self.assertIn(field, schema['properties']) self.assertEqual('array', schema['properties'][field]['type']) self.assertEqual( 'string', schema['properties'][field]['items']['type'], ) self.assertEqual(100, schema['properties']['limit']['maximum']) def test_metadata_guides_ai_to_explicit_fields_without_fallback(self): metadata = QueryOrderExactTool().metadata() description = metadata['description'] properties = metadata['input_schema']['properties'] for phrase in ( '单号类型不明确', '先询问用户', '不得改用其他字段', '同一字段使用 IN', '不同字段使用 AND', ): self.assertIn(phrase, description) single_fields = ( 'order_number', 'reference_number', 'tracking_number', 'outbound_number', 'container_code', 'so_number', ) batch_fields = ( 'order_numbers', 'reference_numbers', 'tracking_numbers', 'outbound_numbers', 'container_codes', 'so_numbers', ) for field in single_fields: self.assertTrue(properties[field]['description']) self.assertIsInstance(properties[field]['examples'][0], str) for field in batch_fields: self.assertTrue(properties[field]['description']) self.assertIsInstance(properties[field]['examples'][0], list) self.assertGreater(len(properties[field]['examples'][0]), 1) self.assertIn('系统订单号', properties['order_number']['description']) self.assertIn( '客户参考号', properties['reference_number']['description'], ) self.assertIn('承运商', properties['tracking_number']['description']) self.assertIn('出库单号', properties['outbound_number']['description']) self.assertIn('柜号', properties['container_code']['description']) self.assertIn('Shipping Order', properties['so_number']['description']) for field in ( 'shipment_id', 'receiver_country', 'product_ids', 'customer_ids', 'sales_id', 'warehouse_ids', 'department_id', 'inbound_date_start', 'inbound_date_end', 'outbound_date_start', 'outbound_date_end', 'page', 'limit', ): self.assertTrue(properties[field]['description']) def test_call_forwards_normalized_number_arrays(self): tool = QueryOrderExactTool(api_client=RecordingApiClient()) self.assertIn('order_numbers', inspect.signature(tool.call).parameters) tool.call( order_number=' A ', order_numbers=[' B ', 'A'], tracking_numbers=['T1', 'T2'], ) self.assertEqual('A', tool.api_client.last_call['payload']['order_number']) self.assertEqual( ['B', 'A'], tool.api_client.last_call['payload']['order_numbers'], ) self.assertEqual( ['T1', 'T2'], tool.api_client.last_call['payload']['tracking_numbers'], ) def test_call_rejects_invalid_number_arrays(self): tool = QueryOrderExactTool(api_client=RecordingApiClient()) self.assertIn('order_numbers', inspect.signature(tool.call).parameters) invalid_values = ('A,B', [1], [''], [' '], ['X' * 101]) for values in invalid_values: with self.subTest(values=values): with self.assertRaises(ValueError): tool.call(order_numbers=values) with self.assertRaisesRegex(ValueError, 'must not exceed 200'): tool.call( order_numbers=['O{0}'.format(i) for i in range(200)], so_numbers=['S1'], ) def test_call_normalizes_exact_filters_and_id_arrays(self): client = RecordingApiClient() tool = QueryOrderExactTool(api_client=client) result = tool.call( order_number=' USC001 ', customer_ids=['9', 9, 0, -1, '12'], warehouse_ids=['-1', '5'], sales_id='7', page=0, limit=200, request_id='rq_exact', ) self.assertEqual({'code': 'MCP_0000'}, result) self.assertEqual('query_order_exact', client.last_call['tool_code']) self.assertEqual('/mcp/tools/queryOrderExact', client.last_call['route_path']) self.assertEqual('rq_exact', client.last_call['request_id']) self.assertEqual({ 'order_number': 'USC001', 'customer_ids': [9, 12], 'sales_id': 7, 'warehouse_ids': [-1, 5], 'page': 1, 'limit': 100, }, client.last_call['payload']) def test_call_requires_one_business_filter(self): tool = QueryOrderExactTool(api_client=RecordingApiClient()) with self.assertRaisesRegex(ValueError, 'at least one exact order filter'): tool.call(page=1, limit=20) def test_call_requires_api_client(self): with self.assertRaisesRegex(RuntimeError, 'api client is required'): QueryOrderExactTool().call(order_number='USC001') def test_cli_forwards_exact_fields_and_id_lists(self): client = RecordingApiClient() output = StringIO() exit_code = GatewayApp(api_client=client).run_cli([ 'call', '--tool', 'query_order_exact', '--order-number', 'USC001', '--customer-ids', '9,12', '--warehouse-ids=-1,5', '--inbound-date-start', '2026-07-01', ], stdout=output) self.assertEqual(0, exit_code) self.assertEqual({ 'order_number': 'USC001', 'customer_ids': [9, 12], 'warehouse_ids': [-1, 5], 'inbound_date_start': '2026-07-01', 'page': 1, 'limit': 20, }, client.last_call['payload']) def test_cli_forwards_batch_number_lists(self): self.assertTrue(hasattr(gateway_app_module, 'parse_string_list')) self.assertEqual( ['A', 'B'], gateway_app_module.parse_string_list(' A, B, A '), ) client = RecordingApiClient() output = StringIO() exit_code = GatewayApp(api_client=client).run_cli([ 'call', '--tool', 'query_order_exact', '--order-numbers', 'A,B,A', '--tracking-numbers', 'T1,T2', ], stdout=output) self.assertEqual(0, exit_code) self.assertEqual(['A', 'B'], client.last_call['payload']['order_numbers']) self.assertEqual( ['T1', 'T2'], client.last_call['payload']['tracking_numbers'], ) if __name__ == '__main__': unittest.main()