mcp/tests/test_destination_trailer_tools.py

import importlib
import io
import unittest

from app import GatewayApp
from public_gateway import PublicGatewayApp
from services.output_presenter import OutputPresenter


PROGRESS_COLUMNS = [
    ('bl_number', '提单号'), ('container_code', '柜号'),
    ('container_type', '柜型'), ('warehouse_name', '仓库名称'),
    ('providers_name', '拖车行'), ('trailer_status', '提柜状态'),
    ('latest_remark', '最新操作备注'), ('is_direct_send', '直送'),
    ('cabinet_type', '整拼类型'), ('vessel_name', '船名航次'),
    ('route', '航线'), ('clearance_check', '清关查验'),
    ('etd', 'ETD'), ('atd', 'ATD'), ('eta', 'ETA'), ('ata', 'ATA'),
    ('loading_port', '起运港'), ('destination_port', '目的港'),
    ('wharf_name', '到港码头'), ('appt_time', 'APPT时间'),
    ('delivery_end_time', '卡车实际派送时间'),
    ('est_pickup_time', '预计提柜时间'), ('pickup_time', '实际提柜时间'),
    ('inbound_date', '到仓时间'), ('container_return_time', '还柜时间'),
    ('pickup_prescription', '提柜时效'), ('return_prescription', '还柜时效'),
    ('wharf_wait_time', '码头等待时间'),     ('amazon_wait_time', '亚马逊等待时间'),
    ('bill_status', '应付费用状态'),
    ('issue_way', '出单方式'),
    ('release_status', '放货状态'),
    ('freight_collect', '海运费到付'),
    ('is_dg', 'DG'),
    ('is_soc', 'SOC'),
    ('is_oneway', 'ONEWAY'),
    ('appointment_method', '预约方式'),
    ('need_appointment_code', '是否需要预约码'),
    ('warehouse_special_remark', '仓库特殊备注'),
]
DO_COLUMNS = [
    ('trailer_number', '拖车单号'), ('outbound_number', '出库单号'),
    ('container_code', '柜号'), ('container_type', '柜型'),
    ('bl_number', '提单号'), ('is_direct_send', '是否直送'),
    ('original_do_file', '原DO单'), ('new_do_file', '新DO单'),
    ('email_subject', '邮件主题'), ('email_from', '邮件发送人'),
    ('received_at', '邮件接收时间'),
]
FILTER_TYPES = [
    '海外提柜类型', '拖车行', '海外仓', '直送柜', '应付费用', '整拼类型', '运输方式',
]
STAGES = [
    '待安排拖车', '待提柜', '待还柜(海)', '已还柜(海)', '已提货(空)', 'DO单制作',
]


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

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

    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 DestinationTrailerToolContractTest(unittest.TestCase):
    def tool_class(self, module_name, class_name):
        return getattr(importlib.import_module('tools.' + module_name), class_name)

    def test_query_schema_requires_stage_and_forbids_identity(self):
        cls = self.tool_class(
            'query_destination_trailer_list', 'QueryDestinationTrailerListTool'
        )
        metadata = cls().metadata()
        schema = metadata['input_schema']
        self.assertEqual('query_destination_trailer_list', metadata['name'])
        self.assertEqual('/mcp/tools/queryDestinationTrailerList', cls.route_path)
        self.assertFalse(schema['additionalProperties'])
        self.assertEqual(['trailer_stage'], schema['required'])
        self.assertEqual(STAGES, schema['properties']['trailer_stage']['enum'])
        self.assertNotIn('default', schema['properties']['trailer_stage'])
        for forbidden in ('company_id', 'admin_id', 'is_super', 'outbound_id'):
            self.assertNotIn(forbidden, schema['properties'])
        self.assertIn('使用场景：', metadata['description'])
        self.assertIn('禁止使用：', metadata['description'])
        self.assertIn('号码类型不明确时必须先询问用户', metadata['description'])
        self.assertIn('不得根据号码格式猜测', metadata['description'])
        self.assertIn('不得跨字段或跨工具试查', metadata['description'])
        self.assertIn('query_outbound_list', metadata['description'])
        self.assertIn('提单号、柜号、日期和其他筛选均为可选', metadata['description'])
        self.assertNotIn('必须再有', metadata['description'])

    def test_query_forwards_numbers_dates_and_rejects_invalid(self):
        cls = self.tool_class(
            'query_destination_trailer_list', 'QueryDestinationTrailerListTool'
        )
        client = RecordingApiClient()
        cls(client).call(
            trailer_stage='待提柜',
            bl_numbers=[' BL-1 ', 'BL-1'],
            container_codes=['CONT-1'],
            providers_id=8,
            warehouse_id=3,
            is_direct_send=0,
            bill_status=10,
            cabinet_type=1,
            shipping_method=2,
            eta_start='2026-07-01',
            eta_end='2026-07-31',
            page=2,
            limit=30,
            request_id='rq_dt',
        )
        self.assertEqual(
            {
                'trailer_stage': '待提柜',
                'bl_numbers': ['BL-1'],
                'container_codes': ['CONT-1'],
                'providers_id': 8,
                'warehouse_id': 3,
                'is_direct_send': 0,
                'bill_status': 10,
                'cabinet_type': 1,
                'shipping_method': 2,
                'eta_start': '2026-07-01',
                'eta_end': '2026-07-31',
                'page': 2,
                'limit': 30,
            },
            client.calls[-1][2],
        )
        self.assertEqual(
            '/mcp/tools/queryDestinationTrailerList',
            client.calls[-1][1],
        )
        cls(client).call(trailer_stage='待提柜')
        self.assertEqual(
            {
                'trailer_stage': '待提柜',
                'page': 1,
                'limit': 20,
            },
            client.calls[-1][2],
        )
        cls(client).call(trailer_stage='待提柜', providers_id=8)
        with self.assertRaises(RuntimeError):
            cls().call(trailer_stage='待提柜')
        invalid = (
            {'trailer_stage': '全部', 'bl_numbers': ['BL-1']},
            {'trailer_stage': '待提柜', 'eta_start': '2026-07-01'},
            {
                'trailer_stage': '待提柜',
                'eta_start': '2026-07-01',
                'eta_end': '2026-08-01',
            },
            {
                'trailer_stage': 'DO单制作',
                'bl_numbers': ['BL-1'],
                'providers_id': 8,
            },
            {
                'trailer_stage': 'DO单制作',
                'eta_start': '2026-07-01',
                'eta_end': '2026-07-02',
            },
        )
        for arguments in invalid:
            with self.subTest(arguments=arguments):
                with self.assertRaises(ValueError):
                    cls(client).call(**arguments)

    def test_filter_schema_and_forwarding(self):
        cls = self.tool_class(
            'list_destination_trailer_filter_options',
            'ListDestinationTrailerFilterOptionsTool',
        )
        metadata = cls().metadata()
        schema = metadata['input_schema']
        self.assertEqual(['filter_type'], schema['required'])
        self.assertEqual(FILTER_TYPES, schema['properties']['filter_type']['enum'])
        self.assertIn('使用场景：', metadata['description'])
        self.assertIn('禁止使用：', metadata['description'])
        client = RecordingApiClient()
        cls(client).call('拖车行', keyword='美西', page=2, limit=10)
        self.assertEqual(
            {
                'filter_type': '拖车行',
                'keyword': '美西',
                'page': 2,
                'limit': 10,
            },
            client.calls[-1][2],
        )
        with self.assertRaises(ValueError):
            cls(client).call('排舱阶段')
        with self.assertRaises(RuntimeError):
            cls().call('海外提柜类型')
        with self.assertRaises(ValueError):
            cls(client).call('海外提柜类型', keyword='x' * 101)
        with self.assertRaises(ValueError):
            cls(client).call('海外提柜类型', page=0)
        with self.assertRaises(ValueError):
            cls(client).call('海外提柜类型', limit=True)

    def test_query_covers_remaining_validation_branches(self):
        cls = self.tool_class(
            'query_destination_trailer_list', 'QueryDestinationTrailerListTool'
        )
        client = RecordingApiClient()
        with self.assertRaises(ValueError):
            cls(client).call(
                trailer_stage='待提柜',
                bl_numbers=['X'] * 201,
            )
        with self.assertRaises(ValueError):
            cls(client).call(
                trailer_stage='待提柜',
                eta_start='2026-07-02',
                eta_end='2026-07-01',
            )
        with self.assertRaises(ValueError):
            cls(client).call(
                trailer_stage='待提柜',
                bl_numbers=['BL-1'],
                bill_status=99,
            )
        with self.assertRaises(ValueError):
            cls(client).call(trailer_stage='待提柜', bl_numbers='BL-1')
        with self.assertRaises(ValueError):
            cls(client).call(trailer_stage='待提柜', bl_numbers=[1])
        with self.assertRaises(ValueError):
            cls(client).call(trailer_stage='待提柜', bl_numbers=[' '])
        with self.assertRaises(ValueError):
            cls(client).call(
                trailer_stage='待提柜',
                eta_start=20260701,
                eta_end='2026-07-02',
            )
        with self.assertRaises(ValueError):
            cls(client).call(
                trailer_stage='待提柜',
                eta_start='2026-99-01',
                eta_end='2026-07-02',
            )
        with self.assertRaises(ValueError):
            cls(client).call(
                trailer_stage='待提柜',
                eta_start='2026-7-01',
                eta_end='2026-07-02',
            )
        with self.assertRaises(ValueError):
            cls(client).call(
                trailer_stage='待提柜',
                bl_numbers=['BL-1'],
                providers_id=True,
            )
        with self.assertRaises(ValueError):
            cls(client).call(
                trailer_stage='待提柜',
                bl_numbers=['BL-1'],
                warehouse_id=0,
            )
        with self.assertRaises(ValueError):
            cls(client).call(
                trailer_stage='待提柜',
                bl_numbers=['BL-1'],
                page=0,
            )
        with self.assertRaises(ValueError):
            cls(client).call(
                trailer_stage='待提柜',
                bl_numbers=['BL-1'],
                limit=True,
            )
        cls(client).call(
            trailer_stage='待安排拖车',
            pickup_time_start='2026-07-01',
            pickup_time_end='2026-07-02',
            providers_id=8,
            bill_status=20,
        )
        cls(client).call(trailer_stage='DO单制作', bl_numbers=['BL-1'])
        with self.assertRaises(ValueError):
            cls(client).call(
                trailer_stage='待提柜',
                eta_start='20260701',
                eta_end='2026-07-02',
            )

    def test_cli_forwards_optional_trailer_filters(self):
        client = RecordingApiClient()
        app = GatewayApp(api_client=client)
        with self.assertRaises(ValueError):
            app.run_cli([
                'call', '--tool', 'query_destination_trailer_list',
            ], stdout=io.StringIO())
        app.run_cli([
            'call', '--tool', 'query_destination_trailer_list',
            '--trailer-stage', '待还柜(海)',
            '--container-codes', 'C1',
            '--providers-id', '8',
            '--warehouse-id', '3',
            '--is-direct-send', '1',
            '--bill-status', '30',
            '--cabinet-type', '2',
            '--shipping-method', '2',
            '--eta-start', '2026-07-01',
            '--eta-end', '2026-07-02',
            '--pickup-time-start', '2026-07-03',
            '--pickup-time-end', '2026-07-04',
            '--container-return-time-start', '2026-07-05',
            '--container-return-time-end', '2026-07-06',
        ], stdout=io.StringIO())
        self.assertEqual(
            {
                'trailer_stage': '待还柜(海)',
                'container_codes': ['C1'],
                'providers_id': 8,
                'warehouse_id': 3,
                'is_direct_send': 1,
                'bill_status': 30,
                'cabinet_type': 2,
                'shipping_method': 2,
                'eta_start': '2026-07-01',
                'eta_end': '2026-07-02',
                'pickup_time_start': '2026-07-03',
                'pickup_time_end': '2026-07-04',
                'container_return_time_start': '2026-07-05',
                'container_return_time_end': '2026-07-06',
                'page': 1,
                'limit': 20,
            },
            client.calls[-1][2],
        )

    def test_presenter_fails_closed_on_unknown_shape(self):
        presenter = OutputPresenter()
        meta = {
            'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_x',
        }
        self.assertTrue(presenter.present('query_destination_trailer_list', {
            'code': 'MCP_0000',
            'data': {'trailer_stage': '待提柜', 'columns': [], 'records': [], 'extra': 1},
            'meta': meta,
        })['is_error'])
        self.assertTrue(presenter.present('query_destination_trailer_list', {
            'code': 'MCP_0000',
            'data': {
                'trailer_stage': '未知',
                'columns': [],
                'records': [],
            },
            'meta': meta,
        })['is_error'])
        self.assertTrue(presenter.present('query_destination_trailer_list', {
            'code': 'MCP_0000',
            'data': {
                'trailer_stage': '待提柜',
                'columns': [{'key': 'bl_number', 'name': '提单号'}],
                'records': [],
            },
            'meta': meta,
        })['is_error'])
        columns = [{'key': key, 'name': name} for key, name in PROGRESS_COLUMNS]
        columns[0] = {'key': 'bl_number', 'name': '错'}
        self.assertTrue(presenter.present('query_destination_trailer_list', {
            'code': 'MCP_0000',
            'data': {
                'trailer_stage': '待提柜',
                'columns': columns,
                'records': [],
            },
            'meta': meta,
        })['is_error'])
        record = {key: 'v' for key, _ in PROGRESS_COLUMNS}
        record['bl_number'] = 1
        self.assertTrue(presenter.present('query_destination_trailer_list', {
            'code': 'MCP_0000',
            'data': {
                'trailer_stage': '待提柜',
                'columns': [
                    {'key': key, 'name': name} for key, name in PROGRESS_COLUMNS
                ],
                'records': [record],
            },
            'meta': meta,
        })['is_error'])

    def test_local_public_registry_cli_and_safe_counts(self):
        local = GatewayApp().registered_tool_names()
        public = PublicGatewayApp(None, None).registered_tool_names()
        self.assertEqual(local, public)
        self.assertEqual(42, len(local))
        self.assertEqual(41, len(OutputPresenter.SAFE_TOOLS))
        self.assertIn('query_destination_trailer_list', local)
        self.assertIn('list_destination_trailer_filter_options', local)

        client = RecordingApiClient()
        app = GatewayApp(api_client=client)
        app.run_cli([
            'call', '--tool', 'query_destination_trailer_list',
            '--trailer-stage', '待提柜', '--bl-numbers', ' BL-1,BL-2 ',
            '--page', '2', '--limit', '30',
        ], stdout=io.StringIO())
        self.assertEqual(
            {
                'trailer_stage': '待提柜',
                'bl_numbers': ['BL-1', 'BL-2'],
                'page': 2,
                'limit': 30,
            },
            client.calls[-1][2],
        )
        with self.assertRaises(ValueError):
            app.run_cli([
                'call', '--tool', 'list_destination_trailer_filter_options',
            ], stdout=io.StringIO())
        app.run_cli([
            'call', '--tool', 'list_destination_trailer_filter_options',
            '--filter-type', '海外提柜类型',
        ], stdout=io.StringIO())
        self.assertEqual(
            'list_destination_trailer_filter_options',
            client.calls[-1][0],
        )

    def test_presenter_locks_thirty_and_eleven_columns(self):
        presenter = OutputPresenter()
        progress_record = {key: 'v' for key, _ in PROGRESS_COLUMNS}
        progress_record['trailer_status'] = '待提柜'
        result = presenter.present('query_destination_trailer_list', {
            'code': 'MCP_0000',
            'data': {
                'trailer_stage': '待提柜',
                'columns': [
                    {'key': key, 'name': name} for key, name in PROGRESS_COLUMNS
                ],
                'records': [progress_record],
            },
            'meta': {
                'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_p',
            },
        })
        self.assertFalse(result['is_error'])
        self.assertEqual(39, len(result['structured_content']['headers']))
        self.assertEqual('待提柜', result['structured_content']['trailer_stage'])

        do_record = {key: 'd' for key, _ in DO_COLUMNS}
        do_record['is_direct_send'] = '否'
        do_record['original_do_file'] = ''
        do_record['new_do_file'] = 'https://files.example/do.pdf'
        do_ok = presenter.present('query_destination_trailer_list', {
            'code': 'MCP_0000',
            'data': {
                'trailer_stage': 'DO单制作',
                'columns': [
                    {'key': key, 'name': name} for key, name in DO_COLUMNS
                ],
                'records': [do_record],
            },
            'meta': {
                'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_d',
            },
        })
        self.assertFalse(do_ok['is_error'])
        self.assertEqual(11, len(do_ok['structured_content']['headers']))

        bad_url = dict(do_record)
        bad_url['new_do_file'] = 'javascript:alert(1)'
        bad = presenter.present('query_destination_trailer_list', {
            'code': 'MCP_0000',
            'data': {
                'trailer_stage': 'DO单制作',
                'columns': [
                    {'key': key, 'name': name} for key, name in DO_COLUMNS
                ],
                'records': [bad_url],
            },
            'meta': {
                'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_b',
            },
        })
        self.assertTrue(bad['is_error'])
        extra = dict(progress_record)
        extra['outbound_number'] = 'OB1'
        extra_result = presenter.present('query_destination_trailer_list', {
            'code': 'MCP_0000',
            'data': {
                'trailer_stage': '待提柜',
                'columns': [
                    {'key': key, 'name': name} for key, name in PROGRESS_COLUMNS
                ],
                'records': [extra],
            },
            'meta': {
                'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_e',
            },
        })
        self.assertTrue(extra_result['is_error'])