import importlib import io import json import os import unittest from app import GatewayApp from mcp_protocol import McpProtocolHandler from public_gateway import PublicGatewayApp from services.output_presenter import OutputPresenter LIST_KEYS = [ 'outbound_number', 'direct_send', 'remark', 'cargo_type', 'warehouse_name', 'status', 'so_number', 'container_code', 'seal_number', 'container_type', 'shipping_method', 'total_volume', 'total_weight', 'ship_schedule', 'closing_time', 'est_loading_time', 'operator', 'operation_modes', 'operation_time', ] DETAIL_KEYS = [ 'order_number', 'cargo_type', 'customs_files', 'reference_number', 'customer_name', 'declaration_type', 'merge_declare_number', 'order_remark', 'bl_number', 'container_code', 'customer_service', 'est_inbound_date', 'inbound_date', 'status', 'pieces', 'weight', 'volume', 'pickup_place', 'product_name', 'goods_name', 'closing_time', 'clearance_remark', 'import_clearance_remark', 'delivery_address', 'delivery_type', 'delivery_way', 'channel', 'country_name', 'importer_name', 'vat_type', 'packing_type', 'is_abnormal', 'package_method', 'order_reply', ] class RecordingApiClient: def __init__(self): self.last_call = None def list_enabled_tools(self, request_id=''): return { 'code': 'MCP_0000', 'data': {'tool_codes': [ 'query_outbound_list', 'query_outbound_detail', ]}, } 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': {}} 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 list_enabled_tools(self, token, request_id=''): return { 'code': 'MCP_0000', 'data': {'tool_codes': [ 'query_outbound_list', 'query_outbound_detail', ]}, } def call_tool(self, **kwargs): return {'code': 'MCP_0000', 'data': {}} class OutboundQueryToolTest(unittest.TestCase): def tool_class(self, module_name, class_name): root = os.path.dirname(os.path.dirname(__file__)) self.assertTrue(os.path.exists(os.path.join( root, 'tools', module_name + '.py' ))) return getattr(importlib.import_module('tools.' + module_name), class_name) def test_list_metadata_is_bounded_and_never_accepts_identity_overrides(self): cls = self.tool_class('query_outbound_list', 'QueryOutboundListTool') metadata = cls().metadata() schema = metadata['input_schema'] properties = schema['properties'] self.assertEqual('query_outbound_list', metadata['name']) self.assertFalse(schema['additionalProperties']) for forbidden in ('company_id', 'admin_id', 'is_super', 'outbound_id'): self.assertNotIn(forbidden, properties) for field in ( 'outbound_numbers', 'order_numbers', 'container_codes', 'so_numbers', 'bl_numbers', ): self.assertEqual('array', properties[field]['type']) self.assertEqual(100, properties[field]['maxItems']) self.assertEqual(['outbound_status'], schema['required']) self.assertNotIn('default', properties['outbound_status']) self.assertEqual( [20, 30, 60, 70, 80, 90, 100, 110, 120], properties['outbound_status']['enum'], ) for label in ( '国内排舱', '国内拖车', '出库装柜', '出口报关', '干线运输', '目的国清关', '海外拖车', '海外仓入库', '完成', ): self.assertIn(label, properties['outbound_status']['description']) self.assertNotIn('10、20', properties['outbound_status']['description']) self.assertNotIn('40、45、50、60', properties['outbound_status']['description']) self.assertNotIn('default', properties['shipping_method']) self.assertIn('未指定时查询全部', properties['shipping_method']['description']) self.assertEqual(100, properties['limit']['maximum']) self.assertIn('后台排舱单列表', metadata['description']) self.assertIn('不得展示内部参数名', metadata['description']) def test_list_call_normalizes_and_forwards_only_supported_filters(self): cls = self.tool_class('query_outbound_list', 'QueryOutboundListTool') client = RecordingApiClient() tool = cls(api_client=client) tool.call( outbound_numbers=[' PC001 ', 'PC001', 'PC002'], outbound_status=60, shipping_method=2, trailer_types=[1, 2], page=2, limit=50, request_id='rq_list', ) self.assertEqual('query_outbound_list', client.last_call['tool_code']) self.assertEqual('/mcp/tools/queryOutboundList', client.last_call['route_path']) self.assertEqual({ 'outbound_numbers': ['PC001', 'PC002'], 'outbound_status': 60, 'shipping_method': 2, 'trailer_types': [1, 2], 'page': 2, 'limit': 50, }, client.last_call['payload']) def test_list_call_forwards_complete_optional_filter_contract(self): cls = self.tool_class('query_outbound_list', 'QueryOutboundListTool') client = RecordingApiClient() tool = cls(api_client=client) tool.call( order_numbers=['ORDER001'], outbound_status=20, container_codes=['CONT001'], so_numbers=['SO001'], bl_numbers=['BL001'], warehouse_id=3, is_direct_send=0, declaration_types=[1], clearance_types=[2], closing_time_start='2026-07-01', closing_time_end='2026-07-02', est_loading_time_start='2026-07-03', est_loading_time_end='2026-07-04', create_date_start='2026-07-05', create_date_end='2026-07-06', loading_time_start='2026-07-07', loading_time_end='2026-07-08', ) payload = client.last_call['payload'] self.assertNotIn('shipping_method', payload) self.assertEqual(['ORDER001'], payload['order_numbers']) self.assertEqual(['CONT001'], payload['container_codes']) self.assertEqual(['SO001'], payload['so_numbers']) self.assertEqual(['BL001'], payload['bl_numbers']) self.assertEqual(3, payload['warehouse_id']) self.assertEqual(0, payload['is_direct_send']) self.assertEqual([1], payload['declaration_types']) self.assertEqual([2], payload['clearance_types']) for field in ( 'closing_time_start', 'closing_time_end', 'est_loading_time_start', 'est_loading_time_end', 'create_date_start', 'create_date_end', 'loading_time_start', 'loading_time_end', ): self.assertIn(field, payload) def test_list_call_rejects_invalid_boundary_shapes(self): cls = self.tool_class('query_outbound_list', 'QueryOutboundListTool') with self.assertRaises(RuntimeError): cls().call() client = RecordingApiClient() tool = cls(api_client=client) invalid_calls = ( lambda: tool.call(), lambda: tool.call(outbound_status=20, outbound_numbers='PC001'), lambda: tool.call(outbound_status=20, outbound_numbers=[1]), lambda: tool.call(outbound_status=20, outbound_numbers=[' ']), lambda: tool.call(outbound_status=20, outbound_numbers=[]), lambda: tool.call( outbound_status=20, outbound_numbers=[str(i) for i in range(101)], ), lambda: tool.call(outbound_status=20, trailer_types=[]), lambda: tool.call(outbound_status=20, page=True), lambda: tool.call(outbound_status=20, page='bad'), lambda: tool.call(outbound_status=20, page=0), lambda: tool.call(outbound_status=True), lambda: tool.call(outbound_status='bad'), lambda: tool.call(outbound_status=999), lambda: tool.call(outbound_status=20, closing_time_start='x' * 20), ) for invalid_call in invalid_calls: with self.subTest(invalid_call=invalid_call): with self.assertRaises(ValueError): invalid_call() tool.call(outbound_status=20, trailer_types=[1, 1]) self.assertEqual([1], client.last_call['payload']['trailer_types']) def test_detail_requires_outbound_number_and_caps_page_size_at_twenty(self): cls = self.tool_class('query_outbound_detail', 'QueryOutboundDetailTool') metadata = cls().metadata() schema = metadata['input_schema'] self.assertEqual(['outbound_number'], schema['required']) self.assertEqual(20, schema['properties']['limit']['maximum']) self.assertNotIn('outbound_id', schema['properties']) client = RecordingApiClient() tool = cls(api_client=client) with self.assertRaises(ValueError): tool.call(outbound_number=' ', limit=10) with self.assertRaises(ValueError): tool.call(outbound_number='PC001', limit=21) tool.call( outbound_number=' PC001 ', page=2, limit=20, request_id='rq_detail', ) self.assertEqual('/mcp/tools/queryOutboundDetail', client.last_call['route_path']) self.assertEqual({ 'outbound_number': 'PC001', 'page': 2, 'limit': 20, }, client.last_call['payload']) def test_detail_rejects_missing_client_and_invalid_integer_shapes(self): cls = self.tool_class('query_outbound_detail', 'QueryOutboundDetailTool') with self.assertRaises(RuntimeError): cls().call(outbound_number='PC001') tool = cls(api_client=RecordingApiClient()) invalid_calls = ( lambda: tool.call(outbound_number=1), lambda: tool.call(outbound_number='PC001', page=True), lambda: tool.call(outbound_number='PC001', page='bad'), lambda: tool.call(outbound_number='PC001', page=0), ) for invalid_call in invalid_calls: with self.subTest(invalid_call=invalid_call): with self.assertRaises(ValueError): invalid_call() def test_local_and_public_gateways_register_both_tools(self): local = GatewayApp(api_client=RecordingApiClient()) public = PublicGatewayApp(PublicSessionStore(), PublicApiClient()) for name in ('query_outbound_list', 'query_outbound_detail'): self.assertIn(name, local.registered_tool_names()) self.assertIn(name, public.registered_tool_names()) def test_cli_forwards_outbound_list_filters(self): client = RecordingApiClient() app = GatewayApp(api_client=client) output = io.StringIO() result = app.run_cli([ 'call', '--tool', 'query_outbound_list', '--outbound-numbers', 'PC001,PC002', '--bl-numbers', 'BL001,BL002', '--outbound-status', '120', '--shipping-method', '2', '--warehouse-id', '3', '--is-direct-send', '1', '--trailer-types', '1,2', '--closing-time-start', '2026-07-01', '--page', '2', '--limit', '5', ], stdout=output) self.assertEqual(0, result) self.assertEqual({ 'outbound_numbers': ['PC001', 'PC002'], 'bl_numbers': ['BL001', 'BL002'], 'outbound_status': 120, 'shipping_method': 2, 'warehouse_id': 3, 'is_direct_send': 1, 'trailer_types': [1, 2], 'closing_time_start': '2026-07-01', 'page': 2, 'limit': 5, }, client.last_call['payload']) def test_cli_outbound_list_requires_stage_and_omits_unspecified_shipping_method(self): client = RecordingApiClient() app = GatewayApp(api_client=client) with self.assertRaises(ValueError): app.run_cli([ 'call', '--tool', 'query_outbound_list', ], stdout=io.StringIO()) result = app.run_cli([ 'call', '--tool', 'query_outbound_list', '--outbound-status', '20', ], stdout=io.StringIO()) self.assertEqual(0, result) self.assertEqual({ 'outbound_status': 20, 'page': 1, 'limit': 20, }, client.last_call['payload']) def test_cli_requires_and_forwards_outbound_detail_number(self): client = RecordingApiClient() app = GatewayApp(api_client=client) with self.assertRaises(ValueError): app.run_cli([ 'call', '--tool', 'query_outbound_detail', ], stdout=io.StringIO()) result = app.run_cli([ 'call', '--tool', 'query_outbound_detail', '--outbound-number', 'PC001', '--page', '2', '--limit', '10', ], stdout=io.StringIO()) self.assertEqual(0, result) self.assertEqual({ 'outbound_number': 'PC001', 'page': 2, 'limit': 10, }, client.last_call['payload']) def test_list_presenter_outputs_exact_nineteen_chinese_columns(self): presenter = OutputPresenter() data = { 'summary': '当前页返回 1 张排舱单', 'columns': [ {'key': key, 'name': 'backend_' + key} for key in LIST_KEYS ], 'records': [{key: 'display value' for key in LIST_KEYS}], } result = presenter.present('query_outbound_list', { 'code': 'MCP_0000', 'data': data, 'meta': {'page': 1, 'limit': 20, 'has_more': False}, }) self.assertFalse(result['is_error']) content = result['structured_content'] self.assertEqual(19, len(content['headers'])) self.assertEqual(19, len(content['rows'][0])) serialized = json.dumps(content, ensure_ascii=False) for key in LIST_KEYS: self.assertNotIn(key, serialized) self.assertIn('排舱单号', serialized) self.assertIn('拖报清方式', serialized) def test_detail_presenter_outputs_eleven_item_summary_and_thirty_four_fields(self): presenter = OutputPresenter() summary = { 'bl_number': 'BL001', 'container_code': 'CONT001', 'container_type': '纸箱', 'total_volume': '10', 'total_weight': '20', 'total_pieces': 3, 'sku': 4, 'buy_declaration_count': 1, 'general_declaration_count': 2, 'must_load': '1/5', 'backup_load': '2/5', } detail_record = {key: 'display value' for key in DETAIL_KEYS} detail_record['customs_files'] = [{ 'file_name': 'declaration.pdf', 'file_type': 'pdf', 'file_url': 'https://files.example/declaration.pdf', }] result = presenter.present('query_outbound_detail', { 'code': 'MCP_0000', 'data': { 'summary': summary, 'columns': [ {'key': key, 'name': 'backend_' + key} for key in DETAIL_KEYS ], 'records': [detail_record], }, 'meta': {'page': 1, 'limit': 10, 'has_more': False}, }) self.assertFalse(result['is_error']) content = result['structured_content'] self.assertEqual(11, len(content['summary']['headers'])) self.assertEqual(11, len(content['summary']['row'])) self.assertEqual(34, len(content['details']['headers'])) self.assertEqual(34, len(content['details']['rows'][0])) serialized = json.dumps(content, ensure_ascii=False) for key in list(summary) + DETAIL_KEYS: self.assertNotIn(key, serialized) for key in ('file_name', 'file_type', 'file_url'): self.assertNotIn(key, serialized) self.assertIn('提单号', serialized) self.assertIn('订单号', serialized) self.assertIn('报关资料', serialized) self.assertIn('文件名称', serialized) self.assertIn('文件链接', serialized) def test_detail_presenter_rejects_every_malformed_boundary(self): presenter = OutputPresenter() def valid_data(): summary = { key: '' for key in presenter.OUTBOUND_DETAIL_SUMMARY } record = {key: '' for key in DETAIL_KEYS} record['customs_files'] = [] return { 'summary': summary, 'columns': [{'key': key} for key in DETAIL_KEYS], 'records': [record], } malformed = [] data = valid_data() data['summary'] = [] malformed.append(data) data = valid_data() data['columns'] = 'bad' malformed.append(data) data = valid_data() data['columns'] = [] malformed.append(data) data = valid_data() data['records'] = {} malformed.append(data) data = valid_data() del data['summary']['sku'] malformed.append(data) data = valid_data() data['columns'] = [None] malformed.append(data) data = valid_data() data['columns'] = [{'key': 1}] malformed.append(data) data = valid_data() data['columns'] = [{'key': 'unknown'}] malformed.append(data) data = valid_data() data['records'] = [None] malformed.append(data) data = valid_data() data['records'][0]['customs_files'] = 'bad' malformed.append(data) data = valid_data() data['records'][0]['customs_files'] = [None] malformed.append(data) for data in malformed: with self.subTest(data=data): result = presenter.present('query_outbound_detail', { 'code': 'MCP_0000', 'data': data, }) self.assertTrue(result['is_error']) self.assertEqual( '工具返回格式异常', result['structured_content']['message'], ) def test_detail_presenter_renders_tips_without_pagination_and_none_values(self): presenter = OutputPresenter() summary = {key: '' for key in presenter.OUTBOUND_DETAIL_SUMMARY} record = {key: '' for key in DETAIL_KEYS} record['order_number'] = None record['customs_files'] = [] result = presenter.present('query_outbound_detail', { 'code': 'MCP_0000', 'data': { 'summary': summary, 'columns': [{'key': key} for key in DETAIL_KEYS], 'records': [record], 'tips': ['没有更多数据'], }, }) self.assertFalse(result['is_error']) details = result['structured_content']['details'] self.assertNotIn('pagination', details) self.assertEqual(['没有更多数据'], details['tips']) self.assertEqual('', details['rows'][0][0]) self.assertIn('提示：没有更多数据', result['text']) def test_protocol_safe_error_paths_work_without_trace_request_id(self): response = McpProtocolHandler._tool_exception_response( 1, 'query_outbound_detail', ValueError('outbound_number is required'), ) self.assertNotIn('_meta', response['result']) error = McpProtocolHandler._error_response(2, -32600, 'Invalid Request') self.assertNotIn('data', error['error']) if __name__ == '__main__': unittest.main()