import io import unittest from app import GatewayApp from public_gateway import PublicGatewayApp from services.output_presenter import OutputPresenter from tools.export_pallet_data import ExportPalletDataTool class RecordingApiClient: def __init__(self): self.calls = [] def list_enabled_tools(self, request_id=''): return { 'code': 'MCP_0000', 'data': {'tool_codes': ['export_pallet_data']}, } 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': { 'task_ref': 'mexp_pallet', 'status': 'queued', 'retry_after_seconds': 10, }, } class ExportPalletDataToolTest(unittest.TestCase): def _assert_number_array_schema(self, field_schema, business_label): self.assertEqual('array', field_schema['type']) self.assertEqual(1, field_schema['minItems']) self.assertEqual(200, field_schema['maxItems']) items = field_schema['items'] self.assertEqual('string', items['type']) self.assertEqual(1, items['minLength']) self.assertEqual(100, items['maxLength']) self.assertIn('pattern', items) description = field_schema['description'] self.assertIn(business_label, description) self.assertIn('仅当用户明确', description) self.assertIn('不得放入', description) def test_schema_is_closed_without_ids_or_pagination(self): metadata = ExportPalletDataTool().metadata() schema = metadata['input_schema'] properties = schema['properties'] self.assertEqual('export_pallet_data', metadata['name']) self.assertFalse(schema['additionalProperties']) self.assertEqual( { 'container_codes', 'bl_numbers', 'inbound_time_start', 'inbound_time_end', }, set(properties), ) self.assertNotIn('page', properties) self.assertNotIn('limit', properties) self.assertNotIn('ids', properties) self.assertNotIn('inbound_id', properties) self.assertNotIn('inbound_ids', properties) self.assertNotIn('inbound_numbers', properties) self._assert_number_array_schema(properties['container_codes'], '柜号') self._assert_number_array_schema(properties['bl_numbers'], '提单号') description = metadata['description'] self.assertIn('明确要求导出打托数据', description) self.assertIn('三选一', description) self.assertIn('异步导出', description) self.assertIn('query_export_task', description) self.assertIn('不会在本次调用中等待', description) self.assertIn('禁止在一次调用内轮询', description) self.assertIn('柜号', description) self.assertIn('提单号', description) self.assertIn('海外仓入库时间', description) def test_call_rejects_legacy_inbound_numbers_argument(self): client = RecordingApiClient() tool = ExportPalletDataTool(client) with self.assertRaises(TypeError): tool.call(inbound_numbers=['OLD']) self.assertEqual([], client.calls) def test_call_forwards_container_codes(self): client = RecordingApiClient() result = ExportPalletDataTool(client).call( container_codes=[' CONT-1 ', 'CONT-1', 'CONT-2'], ) self.assertEqual('MCP_0000', result['code']) self.assertEqual( ( 'export_pallet_data', '/mcp/tools/exportPalletData', {'container_codes': ['CONT-1', 'CONT-2']}, 'rq_export_pallet_data', ), client.calls[0], ) def test_call_forwards_bl_numbers(self): client = RecordingApiClient() result = ExportPalletDataTool(client).call( bl_numbers=[' BL-1 ', 'BL-1'], ) self.assertEqual('MCP_0000', result['code']) self.assertEqual( ( 'export_pallet_data', '/mcp/tools/exportPalletData', {'bl_numbers': ['BL-1']}, 'rq_export_pallet_data', ), client.calls[0], ) def test_call_ignores_empty_container_codes_when_bl_numbers_provided(self): client = RecordingApiClient() ExportPalletDataTool(client).call( container_codes=[], bl_numbers=['BL-1'], ) self.assertEqual( ( 'export_pallet_data', '/mcp/tools/exportPalletData', {'bl_numbers': ['BL-1']}, 'rq_export_pallet_data', ), client.calls[0], ) def test_call_ignores_empty_container_codes_when_time_range_provided(self): client = RecordingApiClient() ExportPalletDataTool(client).call( container_codes=[], inbound_time_start='2026-09-01', inbound_time_end='2026-09-07', ) self.assertEqual( { 'inbound_time_start': '2026-09-01', 'inbound_time_end': '2026-09-07', }, client.calls[-1][2], ) def test_call_ignores_empty_bl_numbers_when_container_codes_provided(self): client = RecordingApiClient() ExportPalletDataTool(client).call( container_codes=['CONT-1'], bl_numbers=[], ) self.assertEqual( ( 'export_pallet_data', '/mcp/tools/exportPalletData', {'container_codes': ['CONT-1']}, 'rq_export_pallet_data', ), client.calls[0], ) def test_call_forwards_inbound_time_range(self): client = RecordingApiClient() ExportPalletDataTool(client).call( inbound_time_start='2026-09-01', inbound_time_end='2026-09-07 18:30:00', ) self.assertEqual( { 'inbound_time_start': '2026-09-01', 'inbound_time_end': '2026-09-07 18:30:00', }, client.calls[-1][2], ) def test_call_validation_boundaries(self): tool = ExportPalletDataTool() with self.assertRaisesRegex(RuntimeError, 'api client is required'): tool.call(container_codes=['CONT-1']) client = RecordingApiClient() tool = ExportPalletDataTool(client) with self.assertRaisesRegex(ValueError, 'non-empty list'): tool.call(container_codes='CONT-1') with self.assertRaisesRegex(ValueError, 'must be strings'): tool.call(container_codes=[1]) with self.assertRaisesRegex(ValueError, '1 to 100 chars'): tool.call(container_codes=['']) with self.assertRaisesRegex(ValueError, '1 to 100 chars'): tool.call(container_codes=['x' * 101]) with self.assertRaisesRegex(ValueError, 'at most 200'): tool.call(container_codes=['n{0}'.format(i) for i in range(201)]) with self.assertRaisesRegex(ValueError, 'non-empty list'): tool.call(container_codes=[]) with self.assertRaisesRegex(ValueError, 'non-empty list'): tool.call(bl_numbers=[]) with self.assertRaisesRegex(ValueError, 'non-empty list'): tool.call(bl_numbers='BL-1') with self.assertRaisesRegex(ValueError, 'must be strings'): tool.call(bl_numbers=[1]) with self.assertRaisesRegex(ValueError, '1 to 100 chars'): tool.call(bl_numbers=['x' * 101]) with self.assertRaisesRegex(ValueError, 'at most 200'): tool.call(bl_numbers=['BL-{0}'.format(i) for i in range(201)]) with self.assertRaisesRegex(ValueError, 'cannot mix'): tool.call(container_codes=['CONT-1'], bl_numbers=['BL-1']) with self.assertRaisesRegex(ValueError, 'cannot mix'): tool.call( container_codes=['CONT-1'], inbound_time_start='2026-09-01', inbound_time_end='2026-09-02', ) with self.assertRaisesRegex(ValueError, 'cannot mix'): tool.call( bl_numbers=['BL-1'], inbound_time_start='2026-09-01', inbound_time_end='2026-09-02', ) with self.assertRaisesRegex(ValueError, 'both start and end'): tool.call(inbound_time_start='2026-09-01') with self.assertRaisesRegex(ValueError, 'both start and end'): tool.call(inbound_time_end='2026-09-07') with self.assertRaisesRegex(ValueError, 'within 31 days'): tool.call( inbound_time_start='2026-09-10', inbound_time_end='2026-09-01', ) with self.assertRaisesRegex(ValueError, 'within 31 days'): tool.call( inbound_time_start='2026-09-01', inbound_time_end='2026-10-02', ) with self.assertRaisesRegex(ValueError, 'date or datetime'): tool.call( inbound_time_start='09/01/2026', inbound_time_end='2026-09-02', ) with self.assertRaisesRegex(ValueError, 'provide'): tool.call() ExportPalletDataTool(client).call( inbound_time_start='2026-09-01 00:00:00', inbound_time_end='2026-10-01 23:59:59', ) self.assertEqual( { 'inbound_time_start': '2026-09-01 00:00:00', 'inbound_time_end': '2026-10-01 23:59:59', }, client.calls[-1][2], ) def test_cli_forwards_export_filters(self): client = RecordingApiClient() app = GatewayApp(api_client=client) code = app.run_cli([ 'call', '--tool', 'export_pallet_data', '--container-codes', 'CONT-1, CONT-2', ], stdout=io.StringIO()) self.assertEqual(0, code) self.assertEqual( {'container_codes': ['CONT-1', 'CONT-2']}, client.calls[-1][2], ) code = app.run_cli([ 'call', '--tool', 'export_pallet_data', '--bl-numbers', 'BL-1, BL-2', ], stdout=io.StringIO()) self.assertEqual(0, code) self.assertEqual( {'bl_numbers': ['BL-1', 'BL-2']}, client.calls[-1][2], ) code = app.run_cli([ 'call', '--tool', 'export_pallet_data', '--inbound-time-start', '2026-09-01', '--inbound-time-end', '2026-09-07 18:30:00', ], stdout=io.StringIO()) self.assertEqual(0, code) self.assertEqual( { 'inbound_time_start': '2026-09-01', 'inbound_time_end': '2026-09-07 18:30:00', }, client.calls[-1][2], ) def test_local_and_public_registries_include_export_tool(self): local = GatewayApp().registered_tool_names() public = PublicGatewayApp(None, None).registered_tool_names() self.assertEqual(local, public) self.assertEqual(32, len(local)) self.assertIn('export_pallet_data', local) self.assertEqual(31, len(OutputPresenter.SAFE_TOOLS)) def test_presenter_reuses_queued_export_contract(self): presented = OutputPresenter().present( 'export_pallet_data', { 'code': 'MCP_0000', 'data': { 'task_ref': 'mexp_pallet', 'status': 'queued', 'retry_after_seconds': 10, }, }, ) self.assertFalse(presented['is_error']) task = presented['structured_content']['task'] self.assertEqual('queued', task['status']) self.assertEqual('mexp_pallet', task['task_ref']) self.assertEqual(10, task['retry_after_seconds']) def test_wrong_route_path_fails_and_restore_passes(self): tool = ExportPalletDataTool() original = tool.route_path tool.route_path = '/mcp/tools/exportPalletDataWrong' self.assertNotEqual('/mcp/tools/exportPalletData', tool.route_path) tool.route_path = original self.assertEqual('/mcp/tools/exportPalletData', tool.route_path) if __name__ == '__main__': unittest.main()