mcp/tests/test_export_pallet_data_tool.py

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(34, len(local))
        self.assertIn('export_pallet_data', local)
        self.assertEqual(33, 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()