mcp/tests/test_order_abnormal_tools.py

import importlib
import io
import unittest

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


COLUMNS = [
    ('order_number', '订单号'), ('customer_name', '客户'),
    ('product_name', '物流产品'), ('abnormal_status', '问题件状态'),
    ('abnormal_type', '问题件类型'), ('abnormal_desc', '问题件描述'),
    ('outbound_status', '排舱单状态'), ('inbound_date', '入库时间'),
    ('marker_name', '标记人'),
]
STATUSES = [
    '待处理', '放行待审核', '包裹放行', '退件待审核', '退件驳回', '已退件',
]


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

    def list_enabled_tools(self, request_id=''):
        return {
            'code': 'MCP_0000',
            'data': {
                'tool_codes': [
                    'query_order_abnormal_list',
                    'list_order_abnormal_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 OrderAbnormalToolContractTest(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_status_and_forbids_identity(self):
        cls = self.tool_class(
            'query_order_abnormal_list', 'QueryOrderAbnormalListTool'
        )
        metadata = cls().metadata()
        schema = metadata['input_schema']
        self.assertEqual('query_order_abnormal_list', metadata['name'])
        self.assertEqual('/mcp/tools/queryOrderAbnormalList', cls.route_path)
        self.assertFalse(schema['additionalProperties'])
        self.assertEqual(['abnormal_status'], schema['required'])
        self.assertEqual(STATUSES, schema['properties']['abnormal_status']['enum'])
        self.assertNotIn('default', schema['properties']['abnormal_status'])
        for forbidden in (
            'company_id', 'admin_id', 'is_super', 'customer_id', 'product_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('admin/Order/abnormal', metadata['description'])
        self.assertIn('订单号可选', metadata['description'])

    def test_query_forwards_optional_order_numbers(self):
        cls = self.tool_class(
            'query_order_abnormal_list', 'QueryOrderAbnormalListTool'
        )
        client = RecordingApiClient()
        cls(client).call(
            abnormal_status='包裹放行',
            order_numbers=[' FMS-1 ', 'FMS-1'],
            page=2,
            limit=30,
            request_id='rq_abn',
        )
        self.assertEqual(
            (
                'query_order_abnormal_list',
                '/mcp/tools/queryOrderAbnormalList',
                {
                    'abnormal_status': '包裹放行',
                    'order_numbers': ['FMS-1'],
                    'page': 2,
                    'limit': 30,
                },
                'rq_abn',
            ),
            client.calls[-1],
        )
        cls(client).call(abnormal_status='待处理')
        self.assertEqual(
            {'abnormal_status': '待处理', 'page': 1, 'limit': 20},
            client.calls[-1][2],
        )
        with self.assertRaises(ValueError):
            cls(client).call(abnormal_status='结案')
        with self.assertRaises(RuntimeError):
            cls().call(abnormal_status='待处理')

    def test_filter_schema_and_call(self):
        cls = self.tool_class(
            'list_order_abnormal_filter_options',
            'ListOrderAbnormalFilterOptionsTool',
        )
        metadata = cls().metadata()
        schema = metadata['input_schema']
        self.assertEqual('list_order_abnormal_filter_options', metadata['name'])
        self.assertEqual(
            '/mcp/tools/listOrderAbnormalFilterOptions', cls.route_path
        )
        self.assertEqual(['filter_type'], schema['required'])
        self.assertEqual(['问题件状态'], schema['properties']['filter_type']['enum'])
        self.assertIn('query_order_abnormal_list', 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_order_abnormal_list', 'QueryOrderAbnormalListTool'
        )
        client = RecordingApiClient()
        with self.assertRaises(ValueError):
            cls(client).call(
                abnormal_status='待处理',
                order_numbers=['X%s' % i for i in range(201)],
            )
        with self.assertRaises(ValueError):
            cls(client).call(abnormal_status='待处理', order_numbers='FMS-1')
        with self.assertRaises(ValueError):
            cls(client).call(abnormal_status='待处理', order_numbers=[1])
        with self.assertRaises(ValueError):
            cls(client).call(abnormal_status='待处理', order_numbers=[' '])
        with self.assertRaises(ValueError):
            cls(client).call(
                abnormal_status='待处理',
                order_numbers=['X' * 101],
            )
        with self.assertRaises(ValueError):
            cls(client).call(abnormal_status='待处理', page=0)
        with self.assertRaises(ValueError):
            cls(client).call(abnormal_status='待处理', limit=True)
        with self.assertRaises(ValueError):
            cls(client).call(abnormal_status='待处理', order_numbers=[])
        with self.assertRaises(ValueError):
            cls(client).call(abnormal_status='待处理', page=101)

    def test_cli_forwards_status_and_order_numbers(self):
        client = RecordingApiClient()
        app = GatewayApp(api_client=client)
        with self.assertRaises(ValueError):
            app.run_cli([
                'call', '--tool', 'query_order_abnormal_list',
            ], stdout=io.StringIO())
        app.run_cli([
            'call', '--tool', 'query_order_abnormal_list',
            '--abnormal-status', '退件待审核',
            '--order-numbers', ' FMS-1,FMS-2 ',
        ], stdout=io.StringIO())
        self.assertEqual(
            {
                'abnormal_status': '退件待审核',
                'order_numbers': ['FMS-1', 'FMS-2'],
                'page': 1,
                'limit': 20,
            },
            client.calls[-1][2],
        )
        app.run_cli([
            'call', '--tool', 'query_order_abnormal_list',
            '--abnormal-status', '待处理',
        ], stdout=io.StringIO())
        self.assertEqual(
            {'abnormal_status': '待处理', 'page': 1, 'limit': 20},
            client.calls[-1][2],
        )
        app.run_cli([
            'call', '--tool', 'list_order_abnormal_filter_options',
            '--filter-type', '问题件状态',
        ], stdout=io.StringIO())
        self.assertEqual('list_order_abnormal_filter_options', client.calls[-1][0])
        with self.assertRaises(ValueError):
            app.run_cli([
                'call', '--tool', 'list_order_abnormal_filter_options',
            ], stdout=io.StringIO())

    def test_presenter_renders_nine_columns_and_fails_closed(self):
        presenter = OutputPresenter()
        meta = {
            'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_x',
        }
        columns = [{'key': key, 'name': name} for key, name in COLUMNS]
        record = {key: 'v' for key, _ in COLUMNS}
        record['abnormal_status'] = '待处理'
        ok = presenter.present('query_order_abnormal_list', {
            'code': 'MCP_0000',
            'data': {
                'abnormal_status': '待处理',
                'columns': columns,
                'records': [record],
            },
            'meta': meta,
        })
        self.assertFalse(ok['is_error'])
        self.assertEqual(
            ['订单号', '客户', '物流产品', '问题件状态', '问题件类型',
             '问题件描述', '排舱单状态', '入库时间', '标记人'],
            [item['label'] for item in ok['structured_content']['headers']],
        )

        self.assertTrue(presenter.present('query_order_abnormal_list', {
            'code': 'MCP_0000',
            'data': {
                'abnormal_status': '待处理',
                'columns': [],
                'records': [],
                'extra': 1,
            },
            'meta': meta,
        })['is_error'])
        self.assertTrue(presenter.present('query_order_abnormal_list', {
            'code': 'MCP_0000',
            'data': {
                'abnormal_status': '结案',
                'columns': columns,
                'records': [],
            },
            'meta': meta,
        })['is_error'])
        bad_columns = list(columns)
        bad_columns[0] = {'key': 'order_number', 'name': '错'}
        self.assertTrue(presenter.present('query_order_abnormal_list', {
            'code': 'MCP_0000',
            'data': {
                'abnormal_status': '待处理',
                'columns': bad_columns,
                'records': [],
            },
            'meta': meta,
        })['is_error'])
        bad_record = dict(record)
        bad_record['order_number'] = 1
        self.assertTrue(presenter.present('query_order_abnormal_list', {
            'code': 'MCP_0000',
            'data': {
                'abnormal_status': '待处理',
                'columns': columns,
                'records': [bad_record],
            },
            'meta': meta,
        })['is_error'])
        self.assertTrue(presenter.present('query_order_abnormal_list', {
            'code': 'MCP_0000',
            'data': {
                'abnormal_status': '待处理',
                'columns': columns,
                'records': 'bad',
            },
            'meta': meta,
        })['is_error'])
        extra_record = dict(record)
        extra_record['deal_remark'] = 'x'
        self.assertTrue(presenter.present('query_order_abnormal_list', {
            'code': 'MCP_0000',
            'data': {
                'abnormal_status': '待处理',
                'columns': columns,
                'records': [extra_record],
            },
            'meta': meta,
        })['is_error'])
        self.assertTrue(presenter.present('query_order_abnormal_list', {
            'code': 'MCP_0000',
            'data': {
                'abnormal_status': '待处理',
                'columns': None,
                'records': [],
            },
            'meta': meta,
        })['is_error'])
        self.assertTrue(presenter.present('query_order_abnormal_list', {
            'code': 'MCP_0000',
            'data': {
                'abnormal_status': '待处理',
                'columns': columns,
                'records': [None],
            },
            'meta': meta,
        })['is_error'])
        self.assertTrue(presenter.present('query_order_abnormal_list', {
            'code': 'MCP_0000',
            'data': {
                'abnormal_status': '待处理',
                'columns': columns,
                'records': [],
            },
            'meta': {
                'page': 1, 'limit': 20, 'has_more': False,
                'request_id': 'rq_x', 'total': 1,
            },
        })['is_error'])

        options = presenter.present('list_order_abnormal_filter_options', {
            'code': 'MCP_0000',
            'data': {
                'records': [
                    {'value': '待处理', 'label': '待处理', 'code': ''},
                ],
            },
            'meta': meta,
        })
        self.assertFalse(options['is_error'])

    def test_local_public_registry_includes_both_tools(self):
        local = GatewayApp().registered_tool_names()
        public = PublicGatewayApp(None, None).registered_tool_names()
        self.assertEqual(local, public)
        self.assertEqual(30, len(local))
        self.assertEqual(29, len(OutputPresenter.SAFE_TOOLS))
        self.assertIn('query_order_abnormal_list', local)
        self.assertIn('list_order_abnormal_filter_options', local)