mcp/tests/test_query_order_exact_tool.py

import inspect
import unittest
from io import StringIO

import app as gateway_app_module
from app import GatewayApp, parse_int_list
from public_gateway import PublicGatewayApp
from tools.query_order_exact import QueryOrderExactTool


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

    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'}

    def list_enabled_tools(self):
        return {
            'code': 'MCP_0000',
            'data': {'tool_codes': ['query_order_exact']},
        }


class PublicSessionStore:
    def get(self, gateway_session_id):
        if gateway_session_id == 'GWS_test':
            return {'mcp_token': 'MT_test'}
        return None


class PublicApiClient:
    def list_enabled_tools(self, token):
        return {
            'code': 'MCP_0000',
            'data': {'tool_codes': ['query_order_exact']},
        }


class QueryOrderExactToolTest(unittest.TestCase):
    def test_local_and_public_gateways_register_tool(self):
        local_names = {
            tool['name']
            for tool in GatewayApp(api_client=RecordingApiClient()).list_tools()
        }
        public_names = {
            tool['name']
            for tool in PublicGatewayApp(
                PublicSessionStore(),
                PublicApiClient(),
            ).list_tools('GWS_test')
        }

        self.assertIn('query_order_exact', local_names)
        self.assertIn('query_order_exact', public_names)

    def test_parse_int_list_accepts_comma_separated_ids(self):
        self.assertEqual([9, 12], parse_int_list('9, 12'))
        self.assertEqual([], parse_int_list(''))

    def test_metadata_exposes_all_exact_filters(self):
        schema = QueryOrderExactTool().metadata()['input_schema']

        for field in (
            'order_number', 'reference_number', 'tracking_number',
            'outbound_number', 'container_code', 'so_number', 'shipment_id',
            'receiver_country', 'product_ids', 'customer_ids', 'sales_id',
            'warehouse_ids', 'department_id', 'inbound_date_start',
            'inbound_date_end', 'outbound_date_start', 'outbound_date_end',
            'page', 'limit',
        ):
            self.assertIn(field, schema['properties'])
        for field in (
            'order_numbers', 'reference_numbers', 'tracking_numbers',
            'outbound_numbers', 'container_codes', 'so_numbers',
        ):
            self.assertIn(field, schema['properties'])
            self.assertEqual('array', schema['properties'][field]['type'])
            self.assertEqual(
                'string',
                schema['properties'][field]['items']['type'],
            )
        self.assertEqual(100, schema['properties']['limit']['maximum'])

    def test_metadata_guides_ai_to_explicit_fields_without_fallback(self):
        metadata = QueryOrderExactTool().metadata()
        description = metadata['description']
        properties = metadata['input_schema']['properties']

        for phrase in (
            '单号类型不明确', '先询问用户', '不得改用其他字段',
            '同一字段使用 IN', '不同字段使用 AND', '排舱单号',
            '所有单号格式均为开放格式',
            '只能根据用户明确说出的业务类型',
            '前缀、长度、字符组合或示例',
        ):
            self.assertIn(phrase, description)
        self.assertNotIn('出库单号、柜号', description)

        single_number_fields = (
            'order_number', 'reference_number', 'tracking_number',
            'outbound_number', 'container_code', 'so_number', 'shipment_id',
        )
        batch_fields = (
            'order_numbers', 'reference_numbers', 'tracking_numbers',
            'outbound_numbers', 'container_codes', 'so_numbers',
        )
        for field in single_number_fields:
            self.assertTrue(properties[field]['description'])
            self.assertIsInstance(properties[field]['examples'][0], str)
            self.assertIn('用户明确', properties[field]['description'])
        for field in batch_fields:
            self.assertTrue(properties[field]['description'])
            self.assertIsInstance(properties[field]['examples'][0], list)
            self.assertGreater(len(properties[field]['examples'][0]), 1)
            self.assertIn('用户明确', properties[field]['description'])

        self.assertIn('系统订单号', properties['order_number']['description'])
        self.assertIn(
            '客户参考号',
            properties['reference_number']['description'],
        )
        self.assertIn('承运商', properties['tracking_number']['description'])
        outbound_description = properties['outbound_number']['description']
        self.assertIn('排舱单号', outbound_description)
        self.assertIn('不是海外仓出库单号', outbound_description)
        self.assertNotIn('以 PC 开头', outbound_description)
        self.assertNotIn(
            '每个号码均以 PC 开头',
            properties['outbound_numbers']['description'],
        )
        so_description = properties['so_number']['description']
        self.assertIn('格式不固定', so_description)
        self.assertIn('用户明确', so_description)
        self.assertEqual(
            '97964454',
            properties['so_number']['examples'][0],
        )
        self.assertEqual(
            ['97964454', 'OOLU12345678'],
            properties['so_numbers']['examples'][0],
        )
        self.assertIn('柜号', properties['container_code']['description'])
        self.assertIn('Shipping Order', so_description)

        for field in (
            'receiver_country', 'product_ids', 'customer_ids',
            'sales_id', 'warehouse_ids', 'department_id',
            'inbound_date_start', 'inbound_date_end',
            'outbound_date_start', 'outbound_date_end', 'page', 'limit',
        ):
            self.assertTrue(properties[field]['description'])

    def test_call_forwards_normalized_number_arrays(self):
        tool = QueryOrderExactTool(api_client=RecordingApiClient())
        self.assertIn('order_numbers', inspect.signature(tool.call).parameters)

        tool.call(
            order_number=' A ',
            order_numbers=[' B ', 'A'],
            tracking_numbers=['T1', 'T2'],
        )

        self.assertEqual('A', tool.api_client.last_call['payload']['order_number'])
        self.assertEqual(
            ['B', 'A'],
            tool.api_client.last_call['payload']['order_numbers'],
        )
        self.assertEqual(
            ['T1', 'T2'],
            tool.api_client.last_call['payload']['tracking_numbers'],
        )

    def test_call_rejects_invalid_number_arrays(self):
        tool = QueryOrderExactTool(api_client=RecordingApiClient())
        self.assertIn('order_numbers', inspect.signature(tool.call).parameters)

        invalid_values = ('A,B', [1], [''], [' '], ['X' * 101])
        for values in invalid_values:
            with self.subTest(values=values):
                with self.assertRaises(ValueError):
                    tool.call(order_numbers=values)

        with self.assertRaisesRegex(ValueError, 'must not exceed 200'):
            tool.call(
                order_numbers=['O{0}'.format(i) for i in range(200)],
                so_numbers=['S1'],
            )

    def test_call_normalizes_exact_filters_and_id_arrays(self):
        client = RecordingApiClient()
        tool = QueryOrderExactTool(api_client=client)

        result = tool.call(
            order_number=' USC001 ',
            customer_ids=['9', 9, 0, -1, '12'],
            warehouse_ids=['-1', '5'],
            sales_id='7',
            page=0,
            limit=200,
            request_id='rq_exact',
        )

        self.assertEqual({'code': 'MCP_0000'}, result)
        self.assertEqual('query_order_exact', client.last_call['tool_code'])
        self.assertEqual('/mcp/tools/queryOrderExact', client.last_call['route_path'])
        self.assertEqual('rq_exact', client.last_call['request_id'])
        self.assertEqual({
            'order_number': 'USC001',
            'customer_ids': [9, 12],
            'sales_id': 7,
            'warehouse_ids': [-1, 5],
            'page': 1,
            'limit': 100,
        }, client.last_call['payload'])

    def test_call_requires_one_business_filter(self):
        tool = QueryOrderExactTool(api_client=RecordingApiClient())

        with self.assertRaisesRegex(ValueError, 'at least one exact order filter'):
            tool.call(page=1, limit=20)

    def test_call_requires_api_client(self):
        with self.assertRaisesRegex(RuntimeError, 'api client is required'):
            QueryOrderExactTool().call(order_number='USC001')

    def test_call_rejects_non_positive_scalar_ids(self):
        tool = QueryOrderExactTool(api_client=RecordingApiClient())

        for field in ('sales_id', 'department_id'):
            with self.subTest(field=field):
                with self.assertRaisesRegex(ValueError, 'must be greater than 0'):
                    tool.call(order_number='O1', **{field: 0})

    def test_normalize_int_list_accepts_comma_separated_values(self):
        self.assertEqual(
            [9, 12],
            QueryOrderExactTool._normalize_int_list('9, 12, 9, 0'),
        )

    def test_normalize_string_list_deduplicates_values(self):
        self.assertEqual(
            ['A', 'B'],
            QueryOrderExactTool._normalize_string_list(
                ['A', 'B', 'A'],
                'numbers',
            ),
        )

    def test_cli_forwards_exact_fields_and_id_lists(self):
        client = RecordingApiClient()
        output = StringIO()

        exit_code = GatewayApp(api_client=client).run_cli([
            'call',
            '--tool', 'query_order_exact',
            '--order-number', 'USC001',
            '--customer-ids', '9,12',
            '--warehouse-ids=-1,5',
            '--inbound-date-start', '2026-07-01',
        ], stdout=output)

        self.assertEqual(0, exit_code)
        self.assertEqual({
            'order_number': 'USC001',
            'customer_ids': [9, 12],
            'warehouse_ids': [-1, 5],
            'inbound_date_start': '2026-07-01',
            'page': 1,
            'limit': 20,
        }, client.last_call['payload'])

    def test_cli_forwards_batch_number_lists(self):
        self.assertTrue(hasattr(gateway_app_module, 'parse_string_list'))
        self.assertEqual(
            ['A', 'B'],
            gateway_app_module.parse_string_list(' A, B, A '),
        )

        client = RecordingApiClient()
        output = StringIO()
        exit_code = GatewayApp(api_client=client).run_cli([
            'call',
            '--tool', 'query_order_exact',
            '--order-numbers', 'A,B,A',
            '--tracking-numbers', 'T1,T2',
        ], stdout=output)

        self.assertEqual(0, exit_code)
        self.assertEqual(['A', 'B'], client.last_call['payload']['order_numbers'])
        self.assertEqual(
            ['T1', 'T2'],
            client.last_call['payload']['tracking_numbers'],
        )


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