mcp/tests/test_container_timeliness_tools.py

import io
import unittest

from app import GatewayApp
from public_gateway import PublicGatewayApp
from services.output_presenter import OutputPresenter
from tools.query_container_timeliness_list import (
    COLUMNS,
    QueryContainerTimelinessListTool,
)
from tools.export_container_timeliness_report import (
    ExportContainerTimelinessReportTool,
)


class RecordingApiClient:
    def __init__(self):
        self.calls = []

    def list_enabled_tools(self, request_id=''):
        return {
            'code': 'MCP_0000',
            'data': {
                'tool_codes': [
                    'query_container_timeliness_list',
                    'export_container_timeliness_report',
                ],
            },
        }

    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': {}, 'meta': {}}


class ContainerTimelinessToolContractTest(unittest.TestCase):
    def test_query_schema_is_closed_without_identity_or_page_filters(self):
        metadata = QueryContainerTimelinessListTool().metadata()
        schema = metadata['input_schema']
        self.assertEqual('query_container_timeliness_list', metadata['name'])
        self.assertEqual(
            '/mcp/tools/queryContainerTimelinessList',
            QueryContainerTimelinessListTool.route_path,
        )
        self.assertFalse(schema['additionalProperties'])
        self.assertEqual([], schema['required'])
        self.assertEqual(
            {
                'container_codes',
                'bl_numbers',
                'departure_time_start',
                'departure_time_end',
                'page',
                'limit',
            },
            set(schema['properties']),
        )
        for forbidden in (
            'company_id', 'admin_id', 'is_super', 'outbound_numbers',
            'ship_company', 'loading_id',
        ):
            self.assertNotIn(forbidden, schema['properties'])
        description = metadata['description']
        self.assertIn('使用场景：', description)
        self.assertIn('禁止使用：', description)
        self.assertIn('admin/Report/outboundReport', description)
        self.assertIn('query_outbound_list', description)
        self.assertIn('干线实际出发时间', description)
        self.assertIn('54列', description)
        self.assertIn('号码类型不明确时必须先询问用户', description)
        self.assertIn('不得根据号码格式猜测', description)
        self.assertIn('不得跨字段或跨工具试查', description)

    def test_query_forwards_numbers_and_optional_departure_window(self):
        client = RecordingApiClient()
        QueryContainerTimelinessListTool(client).call(
            container_codes=['CONT-1', ' CONT-1 '],
            bl_numbers=['BL-1'],
            departure_time_start='2026-09-01',
            departure_time_end='2026-09-30',
            page=2,
            limit=10,
        )
        self.assertEqual(
            (
                'query_container_timeliness_list',
                '/mcp/tools/queryContainerTimelinessList',
                {
                    'container_codes': ['CONT-1'],
                    'bl_numbers': ['BL-1'],
                    'departure_time_start': '2026-09-01',
                    'departure_time_end': '2026-09-30',
                    'page': 2,
                    'limit': 10,
                },
                'rq_query_container_timeliness_list',
            ),
            client.calls[-1],
        )

    def test_query_rejects_empty_and_oversize_window(self):
        tool = QueryContainerTimelinessListTool(RecordingApiClient())
        with self.assertRaisesRegex(ValueError, 'provide'):
            tool.call()
        with self.assertRaisesRegex(ValueError, 'departure time range requires both'):
            tool.call(departure_time_start='2026-09-01')
        with self.assertRaisesRegex(ValueError, 'departure time window'):
            tool.call(
                departure_time_start='2026-08-01',
                departure_time_end='2026-09-01',
            )
        with self.assertRaisesRegex(ValueError, 'is invalid'):
            tool.call(departure_time_start='09/01/2026', departure_time_end='2026-09-02')
        with self.assertRaisesRegex(ValueError, 'at most 200'):
            tool.call(
                container_codes=['C1'],
                bl_numbers=['B{0}'.format(i) for i in range(200)],
            )
        with self.assertRaisesRegex(ValueError, 'non-empty'):
            tool.call(container_codes=[])
        with self.assertRaisesRegex(ValueError, 'strings'):
            tool.call(container_codes=[1])
        with self.assertRaisesRegex(ValueError, '1 to 100 chars'):
            tool.call(container_codes=[''])
        with self.assertRaisesRegex(ValueError, 'at most 200 container_codes'):
            tool.call(container_codes=['C{0}'.format(i) for i in range(201)])
        with self.assertRaisesRegex(ValueError, 'page is invalid'):
            tool.call(container_codes=['CONT-1'], page=True)
        with self.assertRaisesRegex(ValueError, 'page is invalid'):
            tool.call(container_codes=['CONT-1'], page=0)
        with self.assertRaisesRegex(ValueError, 'limit is invalid'):
            tool.call(container_codes=['CONT-1'], limit='20')
        with self.assertRaisesRegex(ValueError, '1 to 100 chars'):
            tool.call(container_codes=['x' * 101])
        with self.assertRaisesRegex(ValueError, 'departure time window'):
            tool.call(
                departure_time_start='2026-09-30',
                departure_time_end='2026-09-01',
            )
        with self.assertRaisesRegex(RuntimeError, 'api client is required'):
            QueryContainerTimelinessListTool().call(container_codes=['CONT-1'])
        with self.assertRaisesRegex(RuntimeError, 'api client is required'):
            ExportContainerTimelinessReportTool().call(container_codes=['CONT-1'])

    def test_export_schema_forbids_pagination_and_requires_async_follow_up(self):
        metadata = ExportContainerTimelinessReportTool().metadata()
        schema = metadata['input_schema']
        self.assertEqual('export_container_timeliness_report', metadata['name'])
        self.assertFalse(schema['additionalProperties'])
        self.assertEqual(
            {
                'container_codes',
                'bl_numbers',
                'departure_time_start',
                'departure_time_end',
            },
            set(schema['properties']),
        )
        self.assertNotIn('page', schema['properties'])
        description = metadata['description']
        self.assertIn('明确要求导出', description)
        self.assertIn('柜子时效', description)
        self.assertIn('异步导出任务', description)
        self.assertIn('query_export_task', description)
        self.assertIn('不会在本次调用中等待文件生成', description)
        self.assertIn('query_container_timeliness_list', description)

    def test_export_forwards_locator_without_page(self):
        client = RecordingApiClient()
        ExportContainerTimelinessReportTool(client).call(
            bl_numbers=['BL-1'],
        )
        self.assertEqual(
            {
                'bl_numbers': ['BL-1'],
            },
            client.calls[-1][2],
        )
        self.assertNotIn('page', client.calls[-1][2])

    def test_cli_forwards_query_and_export_filters(self):
        client = RecordingApiClient()
        app = GatewayApp(api_client=client)
        code = app.run_cli([
            'call', '--tool', 'query_container_timeliness_list',
            '--container-codes', 'CONT-1',
            '--departure-time-start', '2026-09-01',
            '--departure-time-end', '2026-09-14',
            '--page', '2',
            '--limit', '10',
        ], stdout=io.StringIO())
        self.assertEqual(0, code)
        self.assertEqual(
            {
                'container_codes': ['CONT-1'],
                'departure_time_start': '2026-09-01',
                'departure_time_end': '2026-09-14',
                'page': 2,
                'limit': 10,
            },
            client.calls[-1][2],
        )

        code = app.run_cli([
            'call', '--tool', 'export_container_timeliness_report',
            '--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],
        )
        self.assertNotIn('page', client.calls[-1][2])

    def test_presenter_accepts_fifty_four_string_columns(self):
        presenter = OutputPresenter()
        columns = [{'key': key, 'name': name} for key, name in COLUMNS]
        record = {key: 'v_{0}'.format(key) for key, _ in COLUMNS}
        ok = presenter.present('query_container_timeliness_list', {
            'code': 'MCP_0000',
            'data': {'columns': columns, 'records': [record]},
            'meta': {
                'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_x',
            },
        })
        self.assertFalse(ok['is_error'])
        self.assertEqual(54, len(ok['structured_content']['headers']))
        self.assertEqual('v_container_code', ok['structured_content']['rows'][0][2])

        numeric = dict(record)
        numeric['total_volume'] = 1.2
        self.assertTrue(presenter.present('query_container_timeliness_list', {
            'code': 'MCP_0000',
            'data': {'columns': columns, 'records': [numeric]},
            'meta': {
                'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_x',
            },
        })['is_error'])
        self.assertTrue(presenter.present('query_container_timeliness_list', {
            'code': 'MCP_0000',
            'data': {'columns': columns, 'records': [], 'extra': 1},
            'meta': {
                'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_x',
            },
        })['is_error'])
        self.assertTrue(presenter.present('query_container_timeliness_list', {
            'code': 'MCP_0000',
            'data': {'columns': columns, 'records': []},
            'meta': {
                'page': 1, 'limit': 20, 'has_more': False,
                'request_id': 'rq_x', 'total': 1,
            },
        })['is_error'])
        bad_columns = list(columns)
        bad_columns[0] = {'key': 'ship_company', 'name': '错'}
        self.assertTrue(presenter.present('query_container_timeliness_list', {
            'code': 'MCP_0000',
            'data': {'columns': bad_columns, 'records': []},
            'meta': {
                'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_x',
            },
        })['is_error'])
        extra = dict(record)
        extra['seq'] = '1'
        self.assertTrue(presenter.present('query_container_timeliness_list', {
            'code': 'MCP_0000',
            'data': {'columns': columns, 'records': [extra]},
            'meta': {
                'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_x',
            },
        })['is_error'])
        self.assertTrue(presenter.present('query_container_timeliness_list', {
            'code': 'MCP_0000',
            'data': {'columns': columns, 'records': 'bad'},
            'meta': {
                'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_x',
            },
        })['is_error'])
        short_columns = columns[:-1]
        self.assertTrue(presenter.present('query_container_timeliness_list', {
            'code': 'MCP_0000',
            'data': {'columns': short_columns, 'records': []},
            'meta': {
                'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_x',
            },
        })['is_error'])
        self.assertTrue(presenter.present('query_container_timeliness_list', {
            'code': 'MCP_0000',
            'data': {'columns': columns, 'records': ['bad']},
            'meta': {
                'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_x',
            },
        })['is_error'])

    def test_local_and_public_registries_include_both_tools(self):
        local = GatewayApp().registered_tool_names()
        public = PublicGatewayApp(None, None).registered_tool_names()
        self.assertEqual(local, public)
        self.assertEqual(34, len(local))
        self.assertEqual(33, len(OutputPresenter.SAFE_TOOLS))
        self.assertIn('query_container_timeliness_list', local)
        self.assertIn('export_container_timeliness_report', local)


if __name__ == '__main__':
    unittest.main()