mcp/tests/test_payable_cost_tools.py

import copy
import importlib
import io
import math
from pathlib import Path
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_payable_cost_list',
                    'list_payable_cost_filter_options',
                    'export_payable_cost_list',
                ],
            },
        }

    def call_tool(self, tool_code, route_path, payload, request_id):
        self.calls.append((tool_code, route_path, payload, request_id))
        if tool_code == 'export_payable_cost_list':
            return {
                'code': 'MCP_0000',
                'data': {
                    'task_ref': 'mexp_payable',
                    'status': 'queued',
                    'retry_after_seconds': 10,
                },
            }
        return {'code': 'MCP_0000', 'data': {}, 'meta': {}}


class PayableCostToolContractTest(unittest.TestCase):
    FILTER_TYPES = (
        '业务类型', '业务节点', '物流商', '费用项',
        '是否生成账单', '是否付款', '核销状态', '单据类型',
    )
    QUERY_PROPERTIES = {
        'business_type', 'order_numbers', 'tracking_numbers',
        'container_codes', 'bl_numbers', 'so_numbers',
        'business_date_start', 'business_date_end',
        'cost_date_start', 'cost_date_end',
        'operation_date_start', 'operation_date_end',
        'business_node_id', 'provider_id', 'cost_type_ids',
        'billing_status', 'payment_status', 'verification_status',
        'document_type', 'page', 'limit',
    }

    def tool_class(self, module_name, class_name):
        path = Path(__file__).parents[1] / 'tools' / (module_name + '.py')
        self.assertTrue(path.is_file(), str(path))
        return getattr(importlib.import_module('tools.' + module_name), class_name)

    def test_query_schema_is_closed_and_locks_required_business_type(self):
        cls = self.tool_class(
            'query_payable_cost_list', 'QueryPayableCostListTool'
        )
        metadata = cls().metadata()
        schema = metadata['input_schema']
        self.assertEqual('query_payable_cost_list', metadata['name'])
        self.assertFalse(schema['additionalProperties'])
        self.assertEqual(self.QUERY_PROPERTIES, set(schema['properties']))
        self.assertIn('business_type', schema['required'])
        self.assertEqual(
            [1, 2, 3, 4, 5, 7],
            schema['properties']['business_type']['enum'],
        )
        self.assertIn('list_payable_cost_filter_options', metadata['description'])
        self.assertIn('禁止', metadata['description'])

    def test_query_call_forwards_five_number_fields_and_three_dates(self):
        cls = self.tool_class(
            'query_payable_cost_list', 'QueryPayableCostListTool'
        )
        client = RecordingApiClient()
        cls(client).call(
            business_type=1,
            order_numbers=[' O-1 '],
            tracking_numbers=[' T-1 '],
            container_codes=[' C-1 '],
            bl_numbers=[' B-1 '],
            so_numbers=[' S-1 '],
            business_date_start='2026-07-01',
            business_date_end='2026-07-31',
            cost_date_start='2026-07-02',
            cost_date_end='2026-07-30',
            operation_date_start='2026-07-03',
            operation_date_end='2026-07-29',
            business_node_id=11,
            provider_id=12,
            cost_type_ids=[13, 14],
            billing_status=0,
            payment_status=-1,
            verification_status=1,
            document_type=0,
            page=2,
            limit=100,
            request_id='rq_payable',
        )
        self.assertEqual(
            (
                'query_payable_cost_list',
                '/mcp/tools/queryPayableCostList',
                {
                    'business_type': 1,
                    'order_numbers': ['O-1'],
                    'tracking_numbers': ['T-1'],
                    'container_codes': ['C-1'],
                    'bl_numbers': ['B-1'],
                    'so_numbers': ['S-1'],
                    'business_date_start': '2026-07-01',
                    'business_date_end': '2026-07-31',
                    'cost_date_start': '2026-07-02',
                    'cost_date_end': '2026-07-30',
                    'operation_date_start': '2026-07-03',
                    'operation_date_end': '2026-07-29',
                    'business_node_id': 11,
                    'provider_id': 12,
                    'cost_type_ids': [13, 14],
                    'billing_status': 0,
                    'payment_status': -1,
                    'verification_status': 1,
                    'document_type': 0,
                    'page': 2,
                    'limit': 100,
                },
                'rq_payable',
            ),
            client.calls[-1],
        )

    def test_query_call_rejects_number_date_and_filter_boundaries(self):
        cls = self.tool_class(
            'query_payable_cost_list', 'QueryPayableCostListTool'
        )
        tool = cls(RecordingApiClient())
        invalid = [
            {},
            {'business_type': 6, 'order_numbers': ['O']},
            {'business_type': 1},
            {'business_type': 1, 'order_numbers': 'O'},
            {'business_type': 1, 'order_numbers': [1]},
            {'business_type': 1, 'order_numbers': ['']},
            {'business_type': 1, 'order_numbers': ['x' * 101]},
            {'business_type': 1, 'order_numbers': [str(i) for i in range(201)]},
            {'business_type': 1, 'order_numbers': ['O'] * 201},
            {'business_type': 1, 'business_date_start': '2026-07-01'},
            {
                'business_type': 1,
                'cost_date_start': '2026-07-01',
                'cost_date_end': '2026-08-01',
            },
            {'business_type': 1, 'operation_date_end': '2026-07-01'},
            {'business_type': 1, 'order_numbers': ['O'], 'cost_type_ids': []},
            {'business_type': 1, 'order_numbers': ['O'], 'cost_type_ids': [True]},
            {'business_type': 1, 'order_numbers': ['O'], 'cost_type_ids': list(range(1, 202))},
            {'business_type': 1, 'order_numbers': ['O'], 'page': 101},
            {'business_type': 1, 'order_numbers': ['O'], 'limit': 0},
        ]
        for arguments in invalid:
            with self.subTest(arguments=arguments):
                with self.assertRaises((TypeError, ValueError)):
                    tool.call(**arguments)

    def test_filter_schema_and_call_lock_eight_types_and_cost_linkage(self):
        cls = self.tool_class(
            'list_payable_cost_filter_options',
            'ListPayableCostFilterOptionsTool',
        )
        metadata = cls().metadata()
        schema = metadata['input_schema']
        self.assertFalse(schema['additionalProperties'])
        self.assertEqual(
            {'filter_type', 'business_type', 'keyword', 'page', 'limit'},
            set(schema['properties']),
        )
        self.assertEqual(
            list(self.FILTER_TYPES),
            schema['properties']['filter_type']['enum'],
        )
        client = RecordingApiClient()
        cls(client).call(
            filter_type=' 费用项 ', business_type=3,
            keyword=' 运 ', page=2, limit=30,
        )
        self.assertEqual(
            {
                'filter_type': '费用项', 'business_type': 3,
                'keyword': '运', 'page': 2, 'limit': 30,
            },
            client.calls[-1][2],
        )
        for arguments in (
            {'filter_type': '费用项'},
            {'filter_type': '业务节点', 'business_type': 6},
            {'filter_type': '未知'},
            {'filter_type': '业务类型', 'keyword': 1},
            {'filter_type': '业务类型', 'page': 0},
        ):
            with self.subTest(arguments=arguments):
                with self.assertRaises((TypeError, ValueError)):
                    cls(client).call(**arguments)

    def test_export_schema_matches_query_without_pagination(self):
        cls = self.tool_class(
            'export_payable_cost_list', 'ExportPayableCostListTool'
        )
        metadata = cls().metadata()
        schema = metadata['input_schema']
        self.assertEqual('export_payable_cost_list', metadata['name'])
        self.assertFalse(schema['additionalProperties'])
        self.assertEqual(
            self.QUERY_PROPERTIES - {'page', 'limit'},
            set(schema['properties']),
        )
        self.assertIn('business_type', schema['required'])
        self.assertIn('query_export_task', metadata['description'])
        self.assertIn('单页签', metadata['description'])

    def test_local_public_registry_cli_and_safe_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_payable_cost_list',
                'list_payable_cost_filter_options',
                'export_payable_cost_list',
            ),
            tuple(name for name in local if 'payable_cost' in name),
        )

        client = RecordingApiClient()
        app = GatewayApp(api_client=client)
        app.run_cli([
            'call', '--tool', 'query_payable_cost_list',
            '--business-type', '1', '--order-numbers', ' O-1,O-2 ',
            '--cost-type-ids', '13,14', '--page', '2', '--limit', '30',
        ], stdout=io.StringIO())
        self.assertEqual(
            {
                'business_type': 1, 'order_numbers': ['O-1', 'O-2'],
                'cost_type_ids': [13, 14], 'page': 2, 'limit': 30,
            },
            client.calls[-1][2],
        )
        app.run_cli([
            'call', '--tool', 'list_payable_cost_filter_options',
            '--filter-type', '费用项', '--business-type', '3',
        ], stdout=io.StringIO())
        self.assertEqual('list_payable_cost_filter_options', client.calls[-1][0])
        app.run_cli([
            'call', '--tool', 'export_payable_cost_list',
            '--business-type', '7', '--order-numbers', 'E-1',
        ], stdout=io.StringIO())
        self.assertEqual('export_payable_cost_list', client.calls[-1][0])

    def test_new_tool_error_paths_and_cli_forwarding_are_covered(self):
        query_cls = self.tool_class(
            'query_payable_cost_list', 'QueryPayableCostListTool'
        )
        with self.assertRaises(RuntimeError):
            query_cls().call(1, order_numbers=['O-1'])
        query = query_cls(RecordingApiClient())
        query.call(business_type=1, order_numbers=['O', 'O'])
        invalid = (
            {
                'business_type': 1, 'business_date_start': '2026-07-02',
                'business_date_end': '2026-07-01',
            },
            {
                'business_type': 1, 'business_date_start': 1,
                'business_date_end': 1,
            },
            {
                'business_type': 1, 'business_date_start': '2026-99-01',
                'business_date_end': '2026-99-02',
            },
            {
                'business_type': 1, 'business_date_start': '20260101',
                'business_date_end': '20260102',
            },
            {'business_type': 1, 'order_numbers': ['O'], 'billing_status': 2},
        )
        for arguments in invalid:
            with self.subTest(arguments=arguments):
                with self.assertRaises((TypeError, ValueError)):
                    query.call(**arguments)

        export_cls = self.tool_class(
            'export_payable_cost_list', 'ExportPayableCostListTool'
        )
        with self.assertRaises(RuntimeError):
            export_cls().call(1, order_numbers=['O-1'])
        with self.assertRaises(ValueError):
            export_cls(RecordingApiClient()).call(6, order_numbers=['O-1'])

        filter_cls = self.tool_class(
            'list_payable_cost_filter_options',
            'ListPayableCostFilterOptionsTool',
        )
        with self.assertRaises(RuntimeError):
            filter_cls().call('业务类型')
        filter_tool = filter_cls(RecordingApiClient())
        with self.assertRaises(ValueError):
            filter_tool.call(1)
        with self.assertRaises(ValueError):
            filter_tool.call('业务类型', business_type=True)
        with self.assertRaises(ValueError):
            filter_tool.call('业务类型', keyword='x' * 101)
        filter_tool.call('业务类型')

        calls = []
        app = GatewayApp()
        app.call_tool = lambda name, arguments, request_id='': (
            calls.append((name, arguments, request_id))
            or {'code': 'MCP_0000', 'data': {}, 'meta': {}}
        )
        output = io.StringIO()
        app.run_cli([
            'call', '--tool', 'query_payable_cost_list', '--business-type', '1',
            '--order-numbers', 'O-1', '--tracking-numbers', 'T-1',
            '--container-codes', 'C-1', '--bl-numbers', 'B-1', '--so-numbers', 'S-1',
            '--business-date-start', '2026-07-01', '--business-date-end', '2026-07-02',
            '--cost-date-start', '2026-07-03', '--cost-date-end', '2026-07-04',
            '--operation-date-start', '2026-07-05', '--operation-date-end', '2026-07-06',
            '--business-node-id', '11', '--provider-id', '12', '--cost-type-ids', '13,14',
            '--billing-status', '1', '--payment-status', '-1',
            '--verification-status', '2', '--document-type', '0',
        ], stdout=output)
        self.assertEqual('query_payable_cost_list', calls[-1][0])
        app.run_cli([
            'call', '--tool', 'export_payable_cost_list', '--business-type', '1',
            '--order-numbers', 'O-1',
        ], stdout=io.StringIO())
        self.assertEqual('export_payable_cost_list', calls[-1][0])
        with self.assertRaises(ValueError):
            app.run_cli(['call', '--tool', 'query_payable_cost_list'], stdout=io.StringIO())
        app.run_cli([
            'call', '--tool', 'list_payable_cost_filter_options',
            '--filter-type', '业务类型',
        ], stdout=io.StringIO())
        app.run_cli([
            'call', '--tool', 'list_payable_cost_filter_options',
            '--filter-type', '业务类型', '--business-type', '3',
        ], stdout=io.StringIO())
        with self.assertRaises(ValueError):
            app.run_cli([
                'call', '--tool', 'list_payable_cost_filter_options'
            ], stdout=io.StringIO())

        for argv in (
            [
                'call', '--tool', 'query_customer_payment_followup',
                '--customer-id', '1', '--department-id', '2', '--sales-id', '3',
                '--merchandiser-id', '4', '--has-unverified-receivable-only', 'false',
            ],
            [
                'call', '--tool', 'query_customer_unverified_bill_details',
                '--customer-id', '1',
            ],
            [
                'call', '--tool', 'query_customer_payment_records',
                '--customer-id', '1', '--receive-date-start', '2026-07-01',
                '--receive-date-end', '2026-07-02',
            ],
            [
                'call', '--tool', 'query_order_receivable_cost_details',
                '--order-number', 'O-1',
            ],
        ):
            app.run_cli(argv, stdout=io.StringIO())
        app.run_cli([
            'call', '--tool', 'query_customer_payment_followup',
        ], stdout=io.StringIO())
        for argv in (
            ['call', '--tool', 'query_customer_unverified_bill_details'],
            ['call', '--tool', 'query_customer_payment_records'],
            ['call', '--tool', 'query_order_receivable_cost_details'],
        ):
            with self.assertRaises(ValueError):
                app.run_cli(argv, stdout=io.StringIO())
        app.run_cli([
            'call', '--tool', 'query_customer_payment_records',
            '--customer-id', '1', '--receive-date-start', '2026-07-01',
        ], stdout=io.StringIO())
        app.run_cli([
            'call', '--tool', 'query_customer_payment_records',
            '--customer-id', '1', '--receive-date-end', '2026-07-02',
        ], stdout=io.StringIO())



class PayableCostPresenterContractTest(unittest.TestCase):
    COLUMN_MAP = {
        1: [
            ('number', '单号'), ('so_number', 'SO号'),
            ('container_code', '柜号'), ('bl_number', '提单号'),
            ('sub_number', '订单号'), ('business_node_name', '业务节点'),
            ('providers_name', '物流商'), ('cost_name', '费用名称'),
            ('payable_cost_q', '金额'), ('currency', '币种'),
            ('company_money_q', '本位币金额'), ('yf_lock', '应付锁定'),
            ('trade_date', '费用发生时间'),
            ('verify_status_text', '核销状态'), ('remark', '备注'),
            ('bill_no', '账单编号'),
            ('payment_order_number', '付款单号'),
        ],
        2: [
            ('number', '单号'), ('sub_number', '跟踪单号'),
            ('workorder_number', '工单号'), ('order_type_name', '单据类型'),
            ('providers_name', '物流商'), ('cost_name', '费用名称'),
            ('payable_cost_q', '应付原币金额'), ('currency', '应付原币币种'),
            ('company_money_q', '本位币金额CNY'), ('yf_lock', '应付锁定'),
            ('business_date', '业务发生时间'),
            ('trade_date', '费用发生时间'), ('create_date', '操作时间'),
            ('verify_status_text', '核销状态'), ('remark', '备注'),
            ('bill_no', '账单编号'),
            ('payment_order_number', '付款单号'),
        ],
        3: [
            ('number', '单号'), ('order_status_text', '包裹状态'),
            ('providers_name', '物流商'),
            ('business_node_name', '费用业务节点'),
            ('cost_name', '费用名称'), ('detail_price', '计费单价'),
            ('charge_weight', '计费重'), ('quote_cost', '计费金额'),
            ('payable_cost_q', '实际应付金额'),
            ('quote_currency', '原币币种'),
            ('company_money_q', '计费本位币金额'),
            ('yf_lock', '应付锁定'), ('bill_no', '账单编号'),
            ('payment_order_number', '付款单号'),
            ('trade_date', '费用发生时间'),
            ('verify_status_text', '核销状态'), ('remark', '财务备注'),
            ('create_user_name', '操作人'), ('create_date', '录入时间'),
        ],
        4: [
            ('number', '订单号'), ('business_bill_no', '货代账单号'),
            ('providers_name', '物流商'), ('cost_name', '费用名称'),
            ('payable_cost_q', '金额'), ('currency', '币种'),
            ('company_money_q', '本位币金额'),
            ('department_name', '事业部'), ('yf_lock', '应付锁定'),
            ('trade_date', '费用发生时间'),
            ('verify_status_text', '核销状态'),
            ('business_node_name', '业务节点'), ('remark', '备注'),
            ('bill_no', '应付账单编号'),
            ('payment_order_number', '付款单号'),
        ],
        5: [
            ('order_type_name', '单据类型'), ('number', '订单号'),
            ('warehouse_name', '仓库'),
            ('business_node_name', '业务节点'),
            ('providers_name', '物流商'), ('cost_name', '费用名称'),
            ('payable_cost_q', '金额'), ('currency', '币种'),
            ('company_money_q', '本位币金额'), ('yf_lock', '应付锁定'),
            ('trade_date', '费用发生时间'),
            ('verify_status_text', '核销状态'), ('remark', '备注'),
            ('bill_no', '账单编号'),
            ('payment_order_number', '付款单号'),
        ],
        7: [
            ('number', '订单号'), ('ep_order_status_text', '订单状态'),
            ('business_node_name', '业务节点'),
            ('providers_name', '物流商'), ('cost_name', '费用名称'),
            ('detail_price', '计费单价'), ('charge_weight', '计费重'),
            ('quote_cost', '计费金额'),
            ('payable_cost_q', '实际应付金额'),
            ('quote_currency', '原币币种'),
            ('company_money_q', '本位币金额'),
            ('trade_date', '费用发生时间'), ('remark', '备注'),
            ('verify_status_text', '核销状态'), ('bill_no', '账单编号'),
            ('payment_order_number', '付款单号'),
            ('create_user_name', '操作人'), ('create_date', '录入时间'),
        ],
    }
    AMOUNT_KEYS = {
        'detail_price', 'charge_weight', 'quote_cost',
        'payable_cost_q', 'company_money_q',
    }

    def payload(self, business_type):
        columns = self.COLUMN_MAP[business_type]
        record = {}
        for key, _ in columns:
            record[key] = 12.5 if key in self.AMOUNT_KEYS else key + '-value'
        return {
            'code': 'MCP_0000',
            'data': {
                'business_type': business_type,
                'company_currency': 'CNY',
                'columns': [
                    {'key': key, 'name': name} for key, name in columns
                ],
                'records': [record],
            },
            'meta': {
                'page': 1, 'limit': 20, 'has_more': False,
                'request_id': 'rq_payable',
            },
        }

    def test_six_business_types_have_exact_dynamic_columns(self):
        presenter = OutputPresenter()
        for business_type, columns in self.COLUMN_MAP.items():
            with self.subTest(business_type=business_type):
                result = presenter.present(
                    'query_payable_cost_list', self.payload(business_type)
                )
                self.assertFalse(result['is_error'])
                self.assertEqual(
                    [{'label': name} for _, name in columns],
                    result['structured_content']['headers'],
                )
                self.assertEqual(
                    len(columns),
                    len(result['structured_content']['rows'][0]),
                )
                self.assertEqual(
                    business_type,
                    result['structured_content']['business_type'],
                )

    def test_unknown_missing_wrong_order_and_non_finite_values_fail_closed(self):
        presenter = OutputPresenter()
        self.assertTrue(presenter.handles('query_payable_cost_list'))
        cases = []
        payload = self.payload(1)
        payload['data']['columns'].append({'key': 'secret', 'name': '秘密'})
        payload['data']['records'][0]['secret'] = 'hidden'
        cases.append(payload)
        payload = self.payload(1)
        payload['data']['columns'].reverse()
        cases.append(payload)
        payload = self.payload(1)
        payload['data']['records'][0].pop('cost_name')
        cases.append(payload)
        payload = self.payload(1)
        payload['data']['records'][0]['secret'] = 'hidden'
        cases.append(payload)
        payload = self.payload(6) if 6 in self.COLUMN_MAP else self.payload(1)
        payload['data']['business_type'] = 6
        cases.append(payload)
        for business_type in self.COLUMN_MAP:
            for key, _ in self.COLUMN_MAP[business_type]:
                if key not in self.AMOUNT_KEYS:
                    continue
                for value in (True, '12.5', math.inf, -math.inf, math.nan):
                    payload = self.payload(business_type)
                    payload['data']['records'][0][key] = value
                    cases.append(payload)
        for payload in cases:
            with self.subTest(payload=payload):
                self.assertTrue(presenter.present(
                    'query_payable_cost_list', payload
                )['is_error'])

    def test_filter_and_export_presenters_use_safe_contracts(self):
        presenter = OutputPresenter()
        filtered = presenter.present(
            'list_payable_cost_filter_options',
            {
                'code': 'MCP_0000',
                'data': {
                    'records': [
                        {'value': 0, 'label': '否', 'code': 'no'},
                        {'value': -1, 'label': '未付款', 'code': 'unpaid'},
                    ],
                },
                'meta': {'page': 1, 'limit': 20, 'has_more': False},
            },
        )
        self.assertFalse(filtered['is_error'])
        self.assertEqual(
            [[0, '否', 'no'], [-1, '未付款', 'unpaid']],
            filtered['structured_content']['rows'],
        )
        exported = presenter.present(
            'export_payable_cost_list',
            {
                'code': 'MCP_0000',
                'data': {
                    'task_ref': 'mexp_payable',
                    'status': 'queued',
                    'retry_after_seconds': 10,
                },
            },
        )
        self.assertFalse(exported['is_error'])
        self.assertEqual(
            'mexp_payable', exported['structured_content']['task']['task_ref']
        )

    def test_payable_filter_presenter_rejects_malformed_records_and_meta(self):
        presenter = OutputPresenter()
        valid = {
            'code': 'MCP_0000',
            'data': {
                'records': [{'value': 0, 'label': 'ok', 'code': 'no'}],
            },
            'meta': {'page': 1, 'limit': 20, 'has_more': False},
        }
        cases = []
        malformed = copy.deepcopy(valid)
        malformed['data']['unexpected'] = True
        cases.append(malformed)
        malformed = copy.deepcopy(valid)
        malformed['data']['records'][0].pop('code')
        cases.append(malformed)
        malformed = copy.deepcopy(valid)
        malformed['data']['records'][0]['value'] = '0'
        cases.append(malformed)
        malformed = copy.deepcopy(valid)
        malformed['data']['records'][0]['label'] = ''
        cases.append(malformed)
        malformed = copy.deepcopy(valid)
        malformed['meta'].pop('has_more')
        cases.append(malformed)
        malformed = copy.deepcopy(valid)
        malformed['meta']['unexpected'] = True
        cases.append(malformed)
        for payload in cases:
            with self.subTest(payload=payload):
                self.assertTrue(
                    presenter.present(
                        'list_payable_cost_filter_options', payload
                    )['is_error']
                )

    def test_presenter_rejects_unknown_top_level_and_non_string_text_cells(self):
        presenter = OutputPresenter()
        payload = self.payload(1)
        payload['data']['unexpected'] = True
        self.assertTrue(
            presenter.present('query_payable_cost_list', payload)['is_error']
        )
        payload = self.payload(1)
        payload['data']['records'][0]['remark'] = 7
        self.assertTrue(
            presenter.present('query_payable_cost_list', payload)['is_error']
        )


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