mcp/tests/test_order_receivable_cost_details_tool.py

import copy
import importlib.util
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):
        self.calls = []

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

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


class OrderReceivableCostDetailsToolTest(unittest.TestCase):
    def load_tool_class(self):
        path = os.path.join(
            os.path.dirname(os.path.dirname(__file__)),
            'tools',
            'query_order_receivable_cost_details.py',
        )
        self.assertTrue(os.path.isfile(path), '订单应收费用 Gateway 工具尚未实现')
        spec = importlib.util.spec_from_file_location('receivable_cost_tool', path)
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)
        return module.QueryOrderReceivableCostDetailsTool

    def test_closed_schema_requires_explicit_order_number_and_forwards_exactly(self):
        tool_class = self.load_tool_class()
        client = RecordingApiClient()
        metadata = tool_class(client).metadata()
        schema = metadata['input_schema']
        self.assertEqual('query_order_receivable_cost_details', metadata['name'])
        self.assertEqual(['order_number'], schema['required'])
        self.assertFalse(schema['additionalProperties'])
        self.assertEqual({'order_number', 'page', 'limit'}, set(schema['properties']))
        self.assertIn('明确订单号', metadata['description'])
        self.assertIn('不得使用参考号', metadata['description'])
        self.assertIn('不得模糊', metadata['description'])

        result = tool_class(client).call(
            order_number='ORD-260001', page=2, limit=30, request_id='rq_cost'
        )
        self.assertEqual('MCP_0000', result['code'])
        self.assertEqual((
            'query_order_receivable_cost_details',
            '/mcp/tools/queryOrderReceivableCostDetails',
            {'order_number': 'ORD-260001', 'page': 2, 'limit': 30},
            'rq_cost',
        ), client.calls[-1])

    def test_invalid_arguments_fail_before_backend_call(self):
        tool_class = self.load_tool_class()
        client = RecordingApiClient()
        tool = tool_class(client)
        for arguments in (
            {}, {'order_number': ''}, {'order_number': '   '},
            {'order_number': 123}, {'order_number': True},
            {'order_number': 'x' * 101}, {'order_number': 'ORD-1', 'page': 0},
            {'order_number': 'ORD-1', 'page': 101},
            {'order_number': 'ORD-1', 'limit': False},
            {'order_number': 'ORD-1', 'limit': 101},
        ):
            with self.subTest(arguments=arguments):
                with self.assertRaises((TypeError, ValueError)):
                    tool.call(**arguments)
        self.assertEqual([], client.calls)
        with self.assertRaises(RuntimeError):
            tool_class().call(order_number='ORD-1')

    def test_registries_cli_and_payment_records_remain_available(self):
        client = RecordingApiClient()
        local = GatewayApp(api_client=client)
        public = PublicGatewayApp(None, None)
        self.assertEqual(local.registered_tool_names(), public.registered_tool_names())
        self.assertEqual(19, len(local.registered_tool_names()))
        self.assertIn('query_order_receivable_cost_details', local.registered_tool_names())
        self.assertIn('query_customer_payment_records', local.registered_tool_names())

        stdout = io.StringIO()
        local.run_cli([
            'call', '--tool', 'query_order_receivable_cost_details',
            '--order-number', 'ORD-260001', '--page', '2', '--limit', '30',
        ], stdout=stdout)
        self.assertEqual('MCP_0000', json.loads(stdout.getvalue())['code'])
        self.assertEqual('ORD-260001', client.calls[-1][2]['order_number'])
        with self.assertRaises(ValueError):
            local.run_cli([
                'call', '--tool', 'query_order_receivable_cost_details',
            ], stdout=io.StringIO())


class OrderReceivableCostDetailsPresenterTest(unittest.TestCase):
    KEYS = [
        'order_number', 'customer_name', 'cost_name', 'cost_customer_name',
        'charge_weight', 'unit_price', 'quote_currency',
        'receivable_original_amount', 'receivable_amount_cny',
        'settlement_amount_cny', 'cost_confirmation_status',
        'closing_status', 'verification_status',
    ]
    HEADERS = [
        '订单号', '客户名称', '费用项', '费用客户', '计费重', '计费单价',
        '原币币种', '应收原币', '应收金额（CNY）', '结算金额（CNY）',
        '费用确认状态', '关账状态', '核销状态',
    ]

    def payload(self):
        return {
            'code': 'MCP_0000',
            'data': {
                'columns': [{'key': key, 'name': key} for key in self.KEYS],
                'records': [{
                    'order_number': 'ORD-260001', 'customer_name': '主客户',
                    'cost_name': '海运费', 'cost_customer_name': '子客户（SC01）',
                    'charge_weight': 12.35, 'unit_price': 8.5,
                    'quote_currency': 'USD', 'receivable_original_amount': 10.0,
                    'receivable_amount_cny': 72.0, 'settlement_amount_cny': 39.0,
                    'cost_confirmation_status': '已确认', 'closing_status': '部分关账',
                    'verification_status': '是',
                }],
            },
            'meta': {'page': 1, 'limit': 20, 'has_more': False},
        }

    def test_presenter_translates_exact_thirteen_columns(self):
        result = OutputPresenter().present(
            'query_order_receivable_cost_details', self.payload()
        )
        self.assertFalse(result['is_error'])
        self.assertEqual(
            [{'label': header} for header in self.HEADERS],
            result['structured_content']['headers'],
        )
        self.assertEqual(13, len(result['structured_content']['rows'][0]))
        serialized = json.dumps(result, ensure_ascii=False)
        for key in self.KEYS:
            self.assertNotIn(key, serialized)

    def test_unknown_missing_reordered_and_bad_values_fail_closed(self):
        cases = []
        unknown = self.payload()
        unknown['data']['records'][0]['secret'] = 'hidden'
        cases.append(unknown)
        missing = self.payload()
        del missing['data']['records'][0]['cost_name']
        cases.append(missing)
        reordered = self.payload()
        reordered['data']['columns'].reverse()
        cases.append(reordered)
        bad_columns = self.payload()
        bad_columns['data']['columns'] = 'bad'
        cases.append(bad_columns)
        bad_column = self.payload()
        bad_column['data']['columns'][0]['internal'] = True
        cases.append(bad_column)
        extra_data = self.payload()
        extra_data['data']['internal'] = True
        cases.append(extra_data)
        bad_meta = self.payload()
        bad_meta['meta']['total'] = 1
        cases.append(bad_meta)
        bad_text = self.payload()
        bad_text['data']['records'][0]['order_number'] = []
        cases.append(bad_text)
        for amount in (True, '10.00', float('inf'), float('-inf'), float('nan')):
            bad_amount = self.payload()
            bad_amount['data']['records'][0]['receivable_amount_cny'] = amount
            cases.append(bad_amount)

        presenter = OutputPresenter()
        for payload in cases:
            with self.subTest(payload=payload):
                self.assertTrue(presenter.present(
                    'query_order_receivable_cost_details', payload
                )['is_error'])

    def test_payload_factory_does_not_share_nested_state(self):
        self.assertEqual(self.payload(), copy.deepcopy(self.payload()))


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