mcp/tests/test_query_customs_declaration_files_tool.py

import importlib
import io
import json
import os
import unittest

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


class RecordingApiClient:
    def __init__(self, enabled=True):
        self.enabled = enabled
        self.last_call = None

    def list_enabled_tools(self, request_id=''):
        return {
            'code': 'MCP_0000',
            'data': {'tool_codes': (
                ['query_customs_declaration_files'] if self.enabled else []
            )},
        }

    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': {'columns': [], 'records': []}}


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 __init__(self, enabled=True):
        self.enabled = enabled
        self.calls = []

    def list_enabled_tools(self, token, request_id=''):
        return {
            'code': 'MCP_0000',
            'data': {'tool_codes': (
                ['query_customs_declaration_files'] if self.enabled else []
            )},
        }

    def call_tool(self, **kwargs):
        self.calls.append(kwargs)
        return {'code': 'MCP_0000'}


class QueryCustomsDeclarationFilesToolTest(unittest.TestCase):
    def tool_class(self):
        path = os.path.join(
            os.path.dirname(os.path.dirname(__file__)),
            'tools',
            'query_customs_declaration_files.py',
        )
        self.assertTrue(os.path.exists(path), path)
        module = importlib.import_module(
            'tools.query_customs_declaration_files'
        )
        return module.QueryCustomsDeclarationFilesTool

    def test_metadata_teaches_ai_exact_order_number_semantics(self):
        metadata = self.tool_class()().metadata()
        schema = metadata['input_schema']
        properties = schema['properties']

        self.assertFalse(schema['additionalProperties'])
        self.assertEqual([
            {'required': ['outbound_numbers']},
            {'required': ['order_numbers']},
        ], schema['oneOf'])
        for field in ('outbound_numbers', 'order_numbers'):
            prop = properties[field]
            self.assertEqual('array', prop['type'])
            self.assertEqual('string', prop['items']['type'])
            self.assertEqual(1, prop['minItems'])
            self.assertEqual(100, prop['maxItems'])
            self.assertNotIn('uniqueItems', prop)

        order_description = properties['order_numbers']['description']
        for phrase in (
            '订单号',
            '后台订单列表',
            '排舱详情',
            '不是系统单号',
            'order_id/id',
            '不是客户参考号',
            '快递单号',
            '排舱单号',
        ):
            self.assertIn(phrase, order_description)
        self.assertNotIn('系统订单号', order_description)

        description = metadata['description']
        for phrase in (
            '用户明确说“订单号”',
            '用户明确说“排舱单号”',
            '没有明确说明是订单号还是排舱单号',
            '必须先提问，让用户选择“订单号”或“排舱单号”',
            '只说“单号”时必须先追问',
            '不是系统单号',
            '用户说“系统单号”时也不得当作订单号',
            '不得根据号码格式猜测',
            '不得跨字段重试',
            '不得展示内部参数名',
        ):
            self.assertIn(phrase, description)
        self.assertNotIn('系统订单号', description)

    def test_outbound_batch_normalizes_and_forwards_supported_fields(self):
        client = RecordingApiClient()
        tool = self.tool_class()(api_client=client)

        result = tool.call(
            outbound_numbers=[' PC001 ', 'PC002', 'PC001'],
            page=2,
            limit=50,
            request_id='rq_customs',
        )

        self.assertEqual('MCP_0000', result['code'])
        self.assertEqual(
            'query_customs_declaration_files',
            client.last_call['tool_code'],
        )
        self.assertEqual(
            '/mcp/tools/queryCustomsDeclarationFiles',
            client.last_call['route_path'],
        )
        self.assertEqual({
            'outbound_numbers': ['PC001', 'PC002'],
            'page': 2,
            'limit': 50,
        }, client.last_call['payload'])

    def test_order_batch_uses_order_numbers_business_field(self):
        client = RecordingApiClient()
        tool = self.tool_class()(api_client=client)

        tool.call(order_numbers=[' ORD001 ', 'ORD002'])

        self.assertEqual({
            'order_numbers': ['ORD001', 'ORD002'],
            'page': 1,
            'limit': 20,
        }, client.last_call['payload'])

    def test_call_rejects_mixed_missing_and_invalid_batches(self):
        tool = self.tool_class()(api_client=RecordingApiClient())
        invalid = (
            {},
            {'outbound_numbers': ['PC001'], 'order_numbers': ['ORD001']},
            {'order_numbers': []},
            {'order_numbers': 'ORD001'},
            {'order_numbers': ['']},
            {'order_numbers': ['ORD001', 2]},
            {'order_numbers': ['X' * 101]},
            {'order_numbers': ['ORD{0}'.format(i) for i in range(101)]},
            {'order_numbers': ['ORD001'], 'page': 0},
            {'order_numbers': ['ORD001'], 'page': 101},
            {'order_numbers': ['ORD001'], 'limit': 101},
            {'order_numbers': ['ORD001'], 'page': True},
            {'order_numbers': ['ORD001'], 'page': 'not-a-number'},
        )
        for arguments in invalid:
            with self.subTest(arguments=arguments):
                with self.assertRaises(ValueError):
                    tool.call(**arguments)

        with self.assertRaisesRegex(RuntimeError, 'api client is required'):
            self.tool_class()().call(order_numbers=['ORD001'])

    def test_local_and_public_gateways_follow_dynamic_registry(self):
        local_enabled = {
            item['name']
            for item in GatewayApp(api_client=RecordingApiClient()).list_tools()
        }
        local_disabled = {
            item['name']
            for item in GatewayApp(
                api_client=RecordingApiClient(enabled=False)
            ).list_tools()
        }
        public_enabled = {
            item['name']
            for item in PublicGatewayApp(
                PublicSessionStore(),
                PublicApiClient(),
            ).list_tools('GWS_test')
        }

        self.assertIn('query_customs_declaration_files', local_enabled)
        self.assertNotIn('query_customs_declaration_files', local_disabled)
        self.assertIn('query_customs_declaration_files', public_enabled)

    def test_public_gateway_forwards_request_scoped_token_without_company_input(self):
        api_client = PublicApiClient()
        gateway = PublicGatewayApp(PublicSessionStore(), api_client)
        self.assertIn(
            'query_customs_declaration_files',
            gateway.registered_tool_names(),
        )

        gateway.call_tool(
            'GWS_test',
            'query_customs_declaration_files',
            {'order_numbers': ['ORD001']},
            request_id='rq_public_customs',
        )

        call = api_client.calls[0]
        self.assertEqual('MT_test', call['token'])
        self.assertEqual(
            '/mcp/tools/queryCustomsDeclarationFiles',
            call['route_path'],
        )
        self.assertEqual({'order_numbers': ['ORD001']}, call['payload'])
        self.assertNotIn('company_id', call['payload'])

    def test_output_presenter_hides_internal_fields_and_keeps_links(self):
        result = OutputPresenter().present(
            'query_customs_declaration_files',
            {
                'code': 'MCP_0000',
                'data': {
                    'summary': '当前页返回 1 个订单的报关资料',
                    'columns': [
                        {'key': 'outbound_number', 'name': '排舱单号'},
                        {'key': 'order_number', 'name': '订单号'},
                        {'key': 'file_name', 'name': '文件名'},
                        {'key': 'file_url', 'name': '文件链接'},
                    ],
                    'records': [{
                        'outbound_number': 'PC001',
                        'order_number': 'ORD001',
                        'file_name': '报关单.pdf',
                        'file_url': 'https://files.test/a.pdf',
                        'order_id': 99,
                    }],
                },
                'meta': {
                    'page': 1,
                    'limit': 20,
                    'has_more': False,
                    'request_id': 'rq_present',
                },
            },
        )

        self.assertFalse(result['is_error'])
        self.assertEqual(
            ['排舱单号', '订单号', '文件名', '文件链接'],
            [header['label'] for header in result['structured_content']['headers']],
        )
        self.assertIn('https://files.test/a.pdf', result['text'])
        serialized = json.dumps(result, ensure_ascii=False)
        for internal in (
            'outbound_number',
            'order_number',
            'file_name',
            'file_url',
            'order_id',
        ):
            self.assertNotIn(internal, serialized)

    def test_cli_forwards_batch_arguments(self):
        client = RecordingApiClient()
        stdout = io.StringIO()
        app = GatewayApp(api_client=client)
        self.assertIn(
            'query_customs_declaration_files',
            app.registered_tool_names(),
        )

        result = app.run_cli([
            'call',
            '--tool', 'query_customs_declaration_files',
            '--order-numbers', 'ORD001,ORD002',
            '--page', '2',
            '--limit', '10',
        ], stdout=stdout)

        self.assertEqual(0, result)
        self.assertEqual({
            'order_numbers': ['ORD001', 'ORD002'],
            'page': 2,
            'limit': 10,
        }, client.last_call['payload'])

        app.run_cli([
            'call',
            '--tool', 'query_customs_declaration_files',
            '--outbound-numbers', 'PC001,PC002',
        ], stdout=io.StringIO())
        self.assertEqual({
            'outbound_numbers': ['PC001', 'PC002'],
            'page': 1,
            'limit': 20,
        }, client.last_call['payload'])


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