mcp/tests/test_receive_volume_tools.py

import importlib
import io
import math
import unittest

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


COLUMNS = [
    ('order_number', '订单号'),
    ('customer_name', '客户名称'),
    ('packing_type', '货物类型'),
    ('inbound_volume', '入库体积'),
    ('inbound_date', '入库时间'),
    ('sales_name', '商务经理'),
    ('merchandiser_name', '客户经理'),
    ('department_name', '事业部'),
    ('product_name', '物流产品'),
]
FILTER_TYPES = ['客户名称', '商务经理', '客户经理', '货物类型', '物流产品']


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

    def list_enabled_tools(self, request_id=''):
        return {
            'code': 'MCP_0000',
            'data': {
                'tool_codes': [
                    'query_receive_volume_list',
                    'list_receive_volume_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 ReceiveVolumeToolContractTest(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_inbound_dates_and_forbids_identity(self):
        cls = self.tool_class(
            'query_receive_volume_list', 'QueryReceiveVolumeListTool'
        )
        metadata = cls().metadata()
        schema = metadata['input_schema']
        self.assertEqual('query_receive_volume_list', metadata['name'])
        self.assertEqual('/mcp/tools/queryReceiveVolumeList', cls.route_path)
        self.assertFalse(schema['additionalProperties'])
        self.assertEqual(
            ['inbound_date_start', 'inbound_date_end'],
            schema['required'],
        )
        for forbidden in (
            'company_id', 'admin_id', 'is_super', 'order_numbers',
            'business_date_start',
        ):
            self.assertNotIn(forbidden, schema['properties'])
        self.assertIn('使用场景：', metadata['description'])
        self.assertIn('禁止使用：', metadata['description'])
        self.assertIn('入库时间', metadata['description'])
        self.assertIn('admin/Report/receiveForm', metadata['description'])
        self.assertIn('list_receive_volume_filter_options', metadata['description'])
        self.assertIn('不得改走query_order', metadata['description'])

    def test_query_forwards_optional_filters(self):
        cls = self.tool_class(
            'query_receive_volume_list', 'QueryReceiveVolumeListTool'
        )
        client = RecordingApiClient()
        cls(client).call(
            inbound_date_start='2026-09-01',
            inbound_date_end='2026-09-30',
            customer_ids=[11, 11, 12],
            sales_ids=[3],
            merchandiser_ids=[4],
            packing_types=[' 散货 ', '散货'],
            product_ids=[8],
            page=2,
            limit=30,
            request_id='rq_rv',
        )
        self.assertEqual(
            (
                'query_receive_volume_list',
                '/mcp/tools/queryReceiveVolumeList',
                {
                    'inbound_date_start': '2026-09-01',
                    'inbound_date_end': '2026-09-30',
                    'customer_ids': [11, 12],
                    'sales_ids': [3],
                    'merchandiser_ids': [4],
                    'packing_types': ['散货'],
                    'product_ids': [8],
                    'page': 2,
                    'limit': 30,
                },
                'rq_rv',
            ),
            client.calls[-1],
        )
        cls(client).call(
            inbound_date_start='2026-09-01',
            inbound_date_end='2026-09-01',
        )
        self.assertEqual(
            {
                'inbound_date_start': '2026-09-01',
                'inbound_date_end': '2026-09-01',
                'page': 1,
                'limit': 20,
            },
            client.calls[-1][2],
        )
        with self.assertRaises(RuntimeError):
            cls().call(
                inbound_date_start='2026-09-01',
                inbound_date_end='2026-09-30',
            )

    def test_query_rejects_invalid_window_and_ids(self):
        cls = self.tool_class(
            'query_receive_volume_list', 'QueryReceiveVolumeListTool'
        )
        client = RecordingApiClient()
        with self.assertRaises(ValueError):
            cls(client).call(
                inbound_date_start='2026-09-30',
                inbound_date_end='2026-09-01',
            )
        with self.assertRaises(ValueError):
            cls(client).call(
                inbound_date_start='2026-08-01',
                inbound_date_end='2026-09-01',
            )
        with self.assertRaises(ValueError):
            cls(client).call(
                inbound_date_start='2026-09-01',
                inbound_date_end='2026-09-30',
                customer_ids=[0],
            )
        with self.assertRaises(ValueError):
            cls(client).call(
                inbound_date_start='2026-09-01',
                inbound_date_end='2026-09-30',
                packing_types=[''],
            )
        with self.assertRaises(ValueError):
            cls(client).call(
                inbound_date_start='2026-09-01',
                inbound_date_end='2026-09-30',
                sales_ids=list(range(1, 202)),
            )
        with self.assertRaises(ValueError):
            cls(client).call(
                inbound_date_start='2026-09-01',
                inbound_date_end='2026-09-30',
                packing_types=['x' * 101],
            )
        with self.assertRaises(ValueError):
            cls(client).call(
                inbound_date_start='2026-09-01',
                inbound_date_end='2026-09-30',
                packing_types=[1],
            )
        with self.assertRaises(ValueError):
            cls(client).call(
                inbound_date_start='2026-09-01',
                inbound_date_end='2026-09-30',
                packing_types=[],
            )
        with self.assertRaises(ValueError):
            cls(client).call(
                inbound_date_start='bad',
                inbound_date_end='2026-09-30',
            )
        with self.assertRaises(ValueError):
            cls(client).call(
                inbound_date_start='2026-9-01',
                inbound_date_end='2026-09-30',
            )
        with self.assertRaises(ValueError):
            cls(client).call(
                inbound_date_start='2026-09-01',
                inbound_date_end='2026-09-30',
                packing_types=['n{0}'.format(i) for i in range(201)],
            )
        with self.assertRaises(ValueError):
            cls(client).call(
                inbound_date_start='2026-09-01',
                inbound_date_end='2026-09-30',
                product_ids=['8'],
            )
        with self.assertRaises(ValueError):
            cls(client).call(
                inbound_date_start='2026-09-01',
                inbound_date_end='2026-09-30',
                merchandiser_ids=[],
            )
        with self.assertRaises(ValueError):
            cls(client).call(
                inbound_date_start='2026-09-01',
                inbound_date_end='2026-09-30',
                page=101,
            )
        with self.assertRaises(ValueError):
            cls(client).call(
                inbound_date_start='2026-09-01',
                inbound_date_end='2026-09-30',
                limit=True,
            )

    def test_filter_schema_and_call(self):
        cls = self.tool_class(
            'list_receive_volume_filter_options',
            'ListReceiveVolumeFilterOptionsTool',
        )
        metadata = cls().metadata()
        schema = metadata['input_schema']
        self.assertEqual('list_receive_volume_filter_options', metadata['name'])
        self.assertEqual(
            '/mcp/tools/listReceiveVolumeFilterOptions', cls.route_path
        )
        self.assertEqual(['filter_type'], schema['required'])
        self.assertEqual(FILTER_TYPES, schema['properties']['filter_type']['enum'])
        self.assertIn('query_receive_volume_list', metadata['description'])
        self.assertIn('禁止猜测', metadata['description'])
        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('客户名称', page=True)

    def test_cli_forwards_dates_and_filters(self):
        client = RecordingApiClient()
        app = GatewayApp(api_client=client)
        with self.assertRaises(ValueError):
            app.run_cli([
                'call', '--tool', 'query_receive_volume_list',
            ], stdout=io.StringIO())
        app.run_cli([
            'call', '--tool', 'query_receive_volume_list',
            '--inbound-date-start', '2026-09-01',
            '--inbound-date-end', '2026-09-30',
            '--customer-ids', '11,12',
            '--sales-ids', '3',
            '--merchandiser-ids', '4',
            '--packing-types', '散货,整柜',
            '--product-ids', '8',
        ], stdout=io.StringIO())
        self.assertEqual(
            {
                'inbound_date_start': '2026-09-01',
                'inbound_date_end': '2026-09-30',
                'customer_ids': [11, 12],
                'sales_ids': [3],
                'merchandiser_ids': [4],
                'packing_types': ['散货', '整柜'],
                'product_ids': [8],
                'page': 1,
                'limit': 20,
            },
            client.calls[-1][2],
        )
        app.run_cli([
            'call', '--tool', 'query_receive_volume_list',
            '--inbound-date-start', '2026-09-01',
            '--inbound-date-end', '2026-09-02',
        ], stdout=io.StringIO())
        self.assertEqual(
            {
                'inbound_date_start': '2026-09-01',
                'inbound_date_end': '2026-09-02',
                'page': 1,
                'limit': 20,
            },
            client.calls[-1][2],
        )
        app.run_cli([
            'call', '--tool', 'list_receive_volume_filter_options',
            '--filter-type', '物流产品',
        ], stdout=io.StringIO())
        self.assertEqual(
            {
                'filter_type': '物流产品',
                'keyword': '',
                'page': 1,
                'limit': 20,
            },
            client.calls[-1][2],
        )

    def test_presenter_accepts_nine_columns_and_finite_volume(self):
        presenter = OutputPresenter()
        columns = [{'key': key, 'name': name} for key, name in COLUMNS]
        record = {
            'order_number': 'FMS-1',
            'customer_name': '客户甲',
            'packing_type': '散货',
            'inbound_volume': 12.5,
            'inbound_date': '2026-09-01',
            'sales_name': '张三',
            'merchandiser_name': '李四',
            'department_name': '事业部A',
            'product_name': '美线-US01',
        }
        meta = {
            'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_x',
        }
        ok = presenter.present('query_receive_volume_list', {
            'code': 'MCP_0000',
            'data': {'columns': columns, 'records': [record]},
            'meta': meta,
        })
        self.assertFalse(ok['is_error'])
        self.assertEqual(12.5, ok['structured_content']['rows'][0][3])

        string_volume = dict(record)
        string_volume['inbound_volume'] = '12.5'
        self.assertTrue(presenter.present('query_receive_volume_list', {
            'code': 'MCP_0000',
            'data': {'columns': columns, 'records': [string_volume]},
            'meta': meta,
        })['is_error'])
        inf_volume = dict(record)
        inf_volume['inbound_volume'] = math.inf
        self.assertTrue(presenter.present('query_receive_volume_list', {
            'code': 'MCP_0000',
            'data': {'columns': columns, 'records': [inf_volume]},
            'meta': meta,
        })['is_error'])
        extra = dict(record)
        extra['inbound_pieces'] = '1'
        self.assertTrue(presenter.present('query_receive_volume_list', {
            'code': 'MCP_0000',
            'data': {'columns': columns, 'records': [extra]},
            'meta': meta,
        })['is_error'])
        self.assertTrue(presenter.present('query_receive_volume_list', {
            'code': 'MCP_0000',
            'data': {'columns': columns, 'records': [], 'extra': 1},
            'meta': meta,
        })['is_error'])
        bad_columns = list(columns)
        bad_columns[0] = {'key': 'order_number', 'name': '错'}
        self.assertTrue(presenter.present('query_receive_volume_list', {
            'code': 'MCP_0000',
            'data': {'columns': bad_columns, 'records': []},
            'meta': meta,
        })['is_error'])
        non_string = dict(record)
        non_string['order_number'] = 1
        self.assertTrue(presenter.present('query_receive_volume_list', {
            'code': 'MCP_0000',
            'data': {'columns': columns, 'records': [non_string]},
            'meta': meta,
        })['is_error'])
        self.assertTrue(presenter.present('query_receive_volume_list', {
            'code': 'MCP_0000',
            'data': {'columns': columns, 'records': []},
            'meta': {
                'page': 1, 'limit': 20, 'has_more': False,
                'request_id': 'rq_x', 'total': 1,
            },
        })['is_error'])

        options = presenter.present('list_receive_volume_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(34, len(local))
        self.assertEqual(33, len(OutputPresenter.SAFE_TOOLS))
        self.assertIn('query_receive_volume_list', local)
        self.assertIn('list_receive_volume_filter_options', local)