mcp/tests/test_receivable_cost_list_tools.py

import copy
import importlib.util
import io
import json
import math
import os
import unittest

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


ROOT = os.path.dirname(os.path.dirname(__file__))


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

    def list_enabled_tools(self, request_id=''):
        return {
            'code': 'MCP_0000',
            'data': {
                'tool_codes': [
                    'query_receivable_cost_list',
                    'list_receivable_cost_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': {}}


class ToolLoaderMixin:
    def load_tool(self, filename, class_name):
        path = os.path.join(ROOT, 'tools', filename)
        self.assertTrue(os.path.isfile(path), filename + ' is not implemented')
        spec = importlib.util.spec_from_file_location(class_name, path)
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)
        return getattr(module, class_name)


class ReceivableCostListToolTest(ToolLoaderMixin, unittest.TestCase):
    NUMBER_FIELDS = (
        'reference_numbers',
        'tracking_numbers',
        'order_numbers',
        'bill_numbers',
    )

    def tool_class(self):
        return self.load_tool(
            'query_receivable_cost_list.py',
            'QueryReceivableCostListTool',
        )

    def test_schema_is_closed_exact_and_zero_inference(self):
        metadata = self.tool_class()().metadata()
        schema = metadata['input_schema']
        self.assertEqual('query_receivable_cost_list', metadata['name'])
        self.assertFalse(schema['additionalProperties'])
        self.assertNotIn('number', schema['properties'])
        self.assertEqual({
            'reference_numbers', 'tracking_numbers', 'order_numbers',
            'bill_numbers', 'business_date_start', 'business_date_end',
            'customer_id', 'sub_customer_id', 'billing_status',
            'verification_status', 'document_type', 'cost_type_id',
            'page', 'limit',
        }, set(schema['properties']))
        for field in self.NUMBER_FIELDS:
            definition = schema['properties'][field]
            self.assertEqual('array', definition['type'])
            self.assertEqual(1, definition['minItems'])
            self.assertEqual(200, definition['maxItems'])
            self.assertEqual({
                'type': 'string', 'minLength': 1, 'maxLength': 100,
                'pattern': '.*\\S.*',
            }, definition['items'])
        self.assertEqual([0, 1], schema['properties']['billing_status']['enum'])
        self.assertEqual(
            [-1, 0, 1],
            schema['properties']['verification_status']['enum'],
        )
        self.assertEqual(0, schema['properties']['document_type']['minimum'])
        self.assertEqual(1, schema['properties']['cost_type_id']['minimum'])
        self.assertIn('禁止根据格式猜测', metadata['description'])
        self.assertIn(
            'list_receivable_cost_filter_options',
            metadata['description'],
        )

    def test_call_forwards_exact_filters_and_date_pair(self):
        tool_class = self.tool_class()
        client = RecordingApiClient()
        result = tool_class(client).call(
            reference_numbers=[' REF-1 '],
            tracking_numbers=['TRACK-1'],
            order_numbers=['ORDER-1'],
            bill_numbers=['BILL-1'],
            business_date_start='2026-07-01',
            business_date_end='2026-07-31',
            customer_id=7,
            sub_customer_id=9,
            billing_status=0,
            verification_status=-1,
            document_type=0,
            cost_type_id=12,
            page=2,
            limit=30,
            request_id='rq_receivable',
        )
        self.assertEqual('MCP_0000', result['code'])
        self.assertEqual((
            'query_receivable_cost_list',
            '/mcp/tools/queryReceivableCostList',
            {
                'reference_numbers': ['REF-1'],
                'tracking_numbers': ['TRACK-1'],
                'order_numbers': ['ORDER-1'],
                'bill_numbers': ['BILL-1'],
                'business_date_start': '2026-07-01',
                'business_date_end': '2026-07-31',
                'customer_id': 7,
                'sub_customer_id': 9,
                'billing_status': 0,
                'verification_status': -1,
                'document_type': 0,
                'cost_type_id': 12,
                'page': 2,
                'limit': 30,
            },
            'rq_receivable',
        ), client.calls[-1])

    def test_call_enforces_cross_field_and_strict_boundaries(self):
        tool = self.tool_class()(RecordingApiClient())
        invalid = [
            {},
            {'business_date_start': '2026-07-01'},
            {'business_date_end': '2026-07-31'},
            {
                'business_date_start': '2026-07-01',
                'business_date_end': '2026-08-01',
            },
            {'reference_numbers': []},
            {'reference_numbers': [7]},
            {'reference_numbers': [' ']},
            {'reference_numbers': ['x' * 101]},
            {'reference_numbers': ['x'] * 201},
            {
                'reference_numbers': ['x'] * 101,
                'tracking_numbers': ['y'] * 100,
            },
            {'reference_numbers': ['x'], 'sub_customer_id': 9},
            {'reference_numbers': ['x'], 'customer_id': True},
            {'reference_numbers': ['x'], 'billing_status': 2},
            {'reference_numbers': ['x'], 'verification_status': 2},
            {'reference_numbers': ['x'], 'document_type': -1},
            {'reference_numbers': ['x'], 'cost_type_id': 0},
            {'reference_numbers': ['x'], 'page': 101},
            {'reference_numbers': ['x'], 'limit': 1.5},
            {
                'reference_numbers': ['x'],
                'business_date_start': 20260701,
                'business_date_end': '2026-07-31',
            },
            {
                'reference_numbers': ['x'],
                'business_date_start': 'not-a-date',
                'business_date_end': '2026-07-31',
            },
            {
                'reference_numbers': ['x'],
                'business_date_start': '20260701',
                'business_date_end': '2026-07-31',
            },
        ]
        for arguments in invalid:
            with self.subTest(arguments=arguments):
                with self.assertRaises((TypeError, ValueError)):
                    tool.call(**arguments)
        self.assertEqual([], tool.api_client.calls)
        with self.assertRaisesRegex(RuntimeError, 'api client is required'):
            self.tool_class()().call(reference_numbers=['REF-1'])


class ReceivableCostFilterOptionsToolTest(ToolLoaderMixin, unittest.TestCase):
    FILTER_TYPES = ['主客户', '子客户', '出账状态', '核销状态', '单据类型', '费用项']

    def tool_class(self):
        return self.load_tool(
            'list_receivable_cost_filter_options.py',
            'ListReceivableCostFilterOptionsTool',
        )

    def test_schema_has_exact_types_and_customer_dependency(self):
        metadata = self.tool_class()().metadata()
        schema = metadata['input_schema']
        self.assertFalse(schema['additionalProperties'])
        self.assertEqual(['filter_type'], schema['required'])
        self.assertEqual(
            self.FILTER_TYPES,
            schema['properties']['filter_type']['enum'],
        )
        self.assertEqual(
            {'type': 'integer', 'minimum': 1},
            schema['properties']['customer_id'],
        )
        self.assertIn('子客户', json.dumps(schema, ensure_ascii=False))
        self.assertIn('禁止猜测', metadata['description'])

    def test_call_forwards_and_enforces_customer_linkage(self):
        tool_class = self.tool_class()
        client = RecordingApiClient()
        tool = tool_class(client)
        tool.call(
            filter_type=' 子客户 ',
            customer_id=7,
            keyword=' 子 ',
            page=2,
            limit=30,
            request_id='rq_filters',
        )
        self.assertEqual((
            'list_receivable_cost_filter_options',
            '/mcp/tools/listReceivableCostFilterOptions',
            {
                'filter_type': '子客户',
                'customer_id': 7,
                'keyword': '子',
                'page': 2,
                'limit': 30,
            },
            'rq_filters',
        ), client.calls[-1])
        tool.call(filter_type='主客户')
        for arguments in (
            {'filter_type': '子客户'},
            {'filter_type': 7},
            {'filter_type': '子客户', 'customer_id': 0},
            {'filter_type': '主客户', 'customer_id': 7},
            {'filter_type': '其他'},
            {'filter_type': '主客户', 'keyword': 7},
            {'filter_type': '主客户', 'keyword': 'x' * 101},
            {'filter_type': '主客户', 'page': 0},
            {'filter_type': '主客户', 'limit': True},
            {'filter_type': '主客户', 'limit': 101},
        ):
            with self.subTest(arguments=arguments):
                with self.assertRaises((TypeError, ValueError)):
                    tool.call(**arguments)
        self.assertEqual(2, len(client.calls))
        with self.assertRaisesRegex(RuntimeError, 'api client is required'):
            tool_class().call('主客户')


class ReceivableCostIntegrationTest(unittest.TestCase):
    def test_registries_are_identical_ordered_and_counts_are_current(self):
        local = GatewayApp().registered_tool_names()
        public = PublicGatewayApp(None, None).registered_tool_names()
        self.assertEqual(local, public)
        self.assertEqual(25, len(local))
        self.assertEqual(24, len(OutputPresenter.SAFE_TOOLS))
        self.assertEqual(
            (
                'query_receivable_cost_list',
                'list_receivable_cost_filter_options',
            ),
            tuple(
                name for name in local
                if name in {
                    'query_receivable_cost_list',
                    'list_receivable_cost_filter_options',
                }
            ),
        )

    def test_cli_forwards_both_tools(self):
        client = RecordingApiClient()
        app = GatewayApp(api_client=client)
        app.run_cli([
            'call', '--tool', 'query_receivable_cost_list',
            '--reference-numbers', ' REF-1,REF-2 ',
            '--bill-numbers', ' BILL-1 ',
            '--business-date-start', '2026-07-01',
            '--business-date-end', '2026-07-31',
            '--customer-id', '7', '--sub-customer-id', '9',
            '--billing-status', '0', '--verification-status', '-1',
            '--document-type', '0', '--cost-type-id', '12',
            '--page', '2', '--limit', '30',
        ], stdout=io.StringIO())
        self.assertEqual({
            'reference_numbers': ['REF-1', 'REF-2'],
            'bill_numbers': ['BILL-1'],
            'business_date_start': '2026-07-01',
            'business_date_end': '2026-07-31',
            'customer_id': 7,
            'sub_customer_id': 9,
            'billing_status': 0,
            'verification_status': -1,
            'document_type': 0,
            'cost_type_id': 12,
            'page': 2,
            'limit': 30,
        }, client.calls[-1][2])
        app.run_cli([
            'call', '--tool', 'query_receivable_cost_list',
            '--order-numbers', 'ORDER-1',
        ], stdout=io.StringIO())
        self.assertEqual({
            'order_numbers': ['ORDER-1'], 'page': 1, 'limit': 20,
        }, client.calls[-1][2])
        app.run_cli([
            'call', '--tool', 'list_receivable_cost_filter_options',
            '--filter-type', '子客户', '--customer-id', '7',
            '--keyword', '子',
        ], stdout=io.StringIO())
        self.assertEqual({
            'filter_type': '子客户', 'customer_id': 7,
            'keyword': '子', 'page': 1, 'limit': 20,
        }, client.calls[-1][2])
        app.run_cli([
            'call', '--tool', 'list_receivable_cost_filter_options',
            '--filter-type', '主客户',
        ], stdout=io.StringIO())
        self.assertEqual({
            'filter_type': '主客户', 'keyword': '',
            'page': 1, 'limit': 20,
        }, client.calls[-1][2])
        with self.assertRaisesRegex(ValueError, '--filter-type is required'):
            app.run_cli([
                'call', '--tool', 'list_receivable_cost_filter_options',
            ], stdout=io.StringIO())


class ReceivableCostPresenterTest(unittest.TestCase):
    KEYS = [
        'main_customer_name', 'sub_customer_name', 'customer_attribute',
        'department_name', 'document_type', 'warehouse_name', 'business_number',
        'reference_number', 'cost_name', 'original_amount', 'total_amount',
        'billed_amount', 'unbilled_amount', 'bill_number', 'verification_status',
        'receipt_number', 'receipt_date', 'sales_name', 'merchandiser_name',
        'drainage_user_name', 'first_leg_order_status', 'settlement_mode',
        'cost_occurred_at', 'business_occurred_at',
    ]
    NAMES = [
        '主客户', '子客户', '客户属性', '事业部', '单据类型', '仓库', '业务单号',
        '参考号', '费用项', '原币金额', '总金额', '已出账金额', '未出账金额',
        '账单编号', '核销状态', '收款水单号', '收款日期', '商务经理',
        '客户经理', '引流人', '头程订单状态', '结算模式', '费用发生时间',
        '业务发生时间',
    ]
    AMOUNT_KEYS = {
        'original_amount', 'total_amount', 'billed_amount', 'unbilled_amount',
    }

    def payload(self):
        record = {key: '值' for key in self.KEYS}
        record.update({
            'original_amount': {'amount': 10.25, 'currency': 'USD'},
            'total_amount': {'amount': 70.5, 'currency': 'CNY'},
            'billed_amount': {'amount': 20.0, 'currency': 'CNY'},
            'unbilled_amount': {'amount': 50.5, 'currency': 'CNY'},
        })
        return {
            'code': 'MCP_0000',
            'data': {
                'columns': [
                    {'key': key, 'name': name}
                    for key, name in zip(self.KEYS, self.NAMES)
                ],
                'records': [record],
            },
            'meta': {
                'page': 1, 'limit': 20, 'has_more': False,
                'request_id': 'rq_receivable',
            },
        }

    def test_query_presenter_requires_exact_contract(self):
        result = OutputPresenter().present(
            'query_receivable_cost_list',
            self.payload(),
        )
        self.assertFalse(result['is_error'])
        self.assertEqual(
            [{'label': name} for name in self.NAMES],
            result['structured_content']['headers'],
        )
        self.assertEqual(24, len(result['structured_content']['rows'][0]))
        self.assertEqual(
            {'page': 1, 'limit': 20, 'has_more': False},
            result['structured_content']['pagination'],
        )

    def test_query_presenter_fails_closed_on_malformed_contract(self):
        cases = []
        for mutation in (
            lambda value: value['data']['columns'].reverse(),
            lambda value: value['data']['columns'][0].update({'secret': True}),
            lambda value: value['data']['columns'][0].update({'name': '错误名称'}),
            lambda value: value['data']['records'][0].update({'secret': 'hidden'}),
            lambda value: value['data']['records'][0].pop('cost_name'),
            lambda value: value['data']['records'][0].update({'cost_name': []}),
            lambda value: value['data'].update({'secret': True}),
            lambda value: value['meta'].update({'total': 1}),
        ):
            payload = self.payload()
            mutation(payload)
            cases.append(payload)
        for key in self.AMOUNT_KEYS:
            for amount in (True, '10', float('inf'), float('-inf'), float('nan')):
                payload = self.payload()
                payload['data']['records'][0][key]['amount'] = amount
                cases.append(payload)
            for malformed in (
                {'amount': 1},
                {'amount': 1, 'currency': 'CNY', 'secret': True},
                {'amount': 1, 'currency': 7},
            ):
                payload = self.payload()
                payload['data']['records'][0][key] = malformed
                cases.append(payload)
        for meta in (
            {'page': 0, 'limit': 20, 'has_more': False},
            {'page': 1, 'limit': 101, 'has_more': False},
            {'page': 1, 'limit': 20, 'has_more': 0},
        ):
            payload = self.payload()
            payload['meta'] = meta
            cases.append(payload)
        presenter = OutputPresenter()
        for payload in cases:
            with self.subTest(payload=payload):
                self.assertTrue(presenter.present(
                    'query_receivable_cost_list', payload
                )['is_error'])

    def test_filter_presenter_preserves_zero_and_negative_values(self):
        payload = {
            'code': 'MCP_0000',
            'data': {
                'records': [
                    {'value': 0, 'label': '未出账', 'code': 'unbilled'},
                    {'value': -1, 'label': '未核销', 'code': 'unverified'},
                ],
            },
            'meta': {'page': 1, 'limit': 20, 'has_more': False},
        }
        result = OutputPresenter().present(
            'list_receivable_cost_filter_options',
            payload,
        )
        self.assertFalse(result['is_error'])
        self.assertEqual(
            [[0, '未出账', 'unbilled'], [-1, '未核销', 'unverified']],
            result['structured_content']['rows'],
        )
        for mutation in (
            lambda value: value['data'].update({'secret': True}),
            lambda value: value['data']['records'][0].update({'secret': True}),
            lambda value: value['data']['records'][0].update({'value': True}),
            lambda value: value['data']['records'][0].update({'label': ''}),
            lambda value: value['data']['records'][0].update({'code': 1}),
            lambda value: value['meta'].update({'total': 2}),
        ):
            malformed = copy.deepcopy(payload)
            mutation(malformed)
            self.assertTrue(OutputPresenter().present(
                'list_receivable_cost_filter_options',
                malformed,
            )['is_error'])


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