mcp/tests/test_outbound_query_tools.py

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()