mcp/tests/test_customer_payment_followup_tool.py

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_customer_payment_followup']},
        }

    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 CustomerPaymentFollowupToolTest(unittest.TestCase):
    def load_tool_class(self):
        path = os.path.join(
            os.path.dirname(os.path.dirname(__file__)),
            'tools',
            'query_customer_payment_followup.py',
        )
        self.assertTrue(os.path.isfile(path), '客户回款跟进 Gateway 工具尚未实现')
        spec = importlib.util.spec_from_file_location('payment_followup_tool', path)
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)
        return module.QueryCustomerPaymentFollowupTool

    def test_closed_schema_and_call_forwarding(self):
        tool_class = self.load_tool_class()
        client = RecordingApiClient()
        tool = tool_class(client)
        metadata = tool.metadata()
        schema = metadata['input_schema']

        self.assertEqual('query_customer_payment_followup', metadata['name'])
        self.assertIn('query_customer_unverified_bill_details', metadata['description'])
        self.assertFalse(schema['additionalProperties'])
        self.assertEqual([], schema['required'])
        self.assertEqual({
            'customer_id', 'department_id', 'sales_id', 'merchandiser_id',
            'has_unverified_receivable_only', 'page', 'limit',
        }, set(schema['properties']))
        self.assertEqual(True, schema['properties']['has_unverified_receivable_only']['default'])
        self.assertEqual(100, schema['properties']['page']['maximum'])
        self.assertEqual(100, schema['properties']['limit']['maximum'])

        result = tool.call(
            customer_id=11,
            department_id=12,
            sales_id=13,
            merchandiser_id=14,
            has_unverified_receivable_only=False,
            page=2,
            limit=40,
            request_id='rq_payment',
        )
        self.assertEqual('MCP_0000', result['code'])
        self.assertEqual((
            'query_customer_payment_followup',
            '/mcp/tools/queryCustomerPaymentFollowup',
            {
                'page': 2,
                'limit': 40,
                'has_unverified_receivable_only': False,
                'customer_id': 11,
                'department_id': 12,
                'sales_id': 13,
                'merchandiser_id': 14,
            },
            'rq_payment',
        ), client.calls[0])

    def test_defaults_and_illegal_parameters_are_rejected(self):
        tool_class = self.load_tool_class()
        client = RecordingApiClient()
        tool = tool_class(client)
        tool.call()
        self.assertEqual({
            'page': 1,
            'limit': 20,
            'has_unverified_receivable_only': True,
        }, client.calls[0][2])

        invalid_calls = (
            {'customer_id': True}, {'department_id': 0}, {'sales_id': 1.5},
            {'merchandiser_id': '2'}, {'page': 0}, {'page': 101},
            {'limit': False}, {'limit': 101},
            {'has_unverified_receivable_only': 1},
            {'has_unverified_receivable_only': 'true'},
        )
        for arguments in invalid_calls:
            with self.subTest(arguments=arguments):
                with self.assertRaises(ValueError):
                    tool.call(**arguments)

        with self.assertRaises(RuntimeError):
            tool_class().call()

    def test_local_public_registries_and_cli_have_eighteen_tools(self):
        local = GatewayApp(api_client=RecordingApiClient())
        public = PublicGatewayApp(None, None)
        self.assertEqual(local.registered_tool_names(), public.registered_tool_names())
        self.assertEqual(25, len(local.registered_tool_names()))
        self.assertIn('query_customer_payment_followup', local.registered_tool_names())

        stdout = io.StringIO()
        local.run_cli([
            'call', '--tool', 'query_customer_payment_followup',
            '--customer-id', '11', '--has-unverified-receivable-only', 'false',
            '--page', '2', '--limit', '10', '--request-id', 'rq_cli_payment',
        ], stdout=stdout)
        self.assertEqual('MCP_0000', json.loads(stdout.getvalue())['code'])
        self.assertEqual({
            'page': 2,
            'limit': 10,
            'customer_id': 11,
            'has_unverified_receivable_only': False,
        }, local.api_client.calls[-1][2])


class CustomerPaymentFollowupPresenterTest(unittest.TestCase):
    KEYS = [
        'customer_name', 'settlement_currency', 'billed_unverified_amount',
        'unbilled_amount', 'overdue_unpaid_amount',
        'billed_unverified_amount_cny', 'unbilled_amount_cny',
        'overdue_unpaid_amount_cny', 'unverified_receivable_monthly_summary',
        'receipt_unverified_amount', 'current_balance', 'credit_limit',
        'bad_debt_total', 'contract_status',
    ]
    SUMMARY_KEYS = [
        'receivable_month', 'unverified_amount', 'is_overdue',
    ]

    def setUp(self):
        self.presenter = OutputPresenter()

    def payload(self):
        record = {
            'customer_name': '甲客户',
            'settlement_currency': 'USD',
            'billed_unverified_amount': 90.0,
            'unbilled_amount': 50.0,
            'overdue_unpaid_amount': 90.0,
            'billed_unverified_amount_cny': 630.0,
            'unbilled_amount_cny': 350.0,
            'overdue_unpaid_amount_cny': 630.0,
            'unverified_receivable_monthly_summary': [{
                'receivable_month': '2026-01',
                'unverified_amount': 90.0,
                'is_overdue': True,
            }],
            'receipt_unverified_amount': 30.0,
            'current_balance': -300.0,
            'credit_limit': 1000.0,
            'bad_debt_total': 15.0,
            'contract_status': '生效中',
        }
        return {
            'code': 'MCP_0000',
            'data': {
                'columns': [{'key': key, 'name': key} for key in self.KEYS],
                'records': [record],
            },
            'meta': {
                'page': 1, 'limit': 20, 'has_more': False,
                'request_id': 'rq_payment',
            },
        }

    def test_success_translates_every_top_level_and_monthly_summary_field(self):
        result = self.presenter.present(
            'query_customer_payment_followup', self.payload()
        )
        self.assertFalse(result['is_error'])
        self.assertEqual(14, len(result['structured_content']['headers']))
        self.assertEqual(14, len(result['structured_content']['rows'][0]))
        summaries = result['structured_content']['rows'][0][8]
        self.assertEqual({
            '应收月份', '未核销金额', '是否逾期',
        }, set(summaries[0]))
        self.assertEqual(
            {'page': 1, 'limit': 20, 'has_more': False},
            result['structured_content']['pagination'],
        )
        instruction = result['structured_content']['display_rules']['instruction']
        self.assertIn('月份汇总', instruction)
        self.assertNotIn('账单明细', instruction)
        serialized = json.dumps(result, ensure_ascii=False)
        for raw_key in self.KEYS + self.SUMMARY_KEYS:
            self.assertNotIn(raw_key, serialized)

    def test_missing_extra_or_reordered_top_level_fields_fail_closed(self):
        cases = []
        missing = self.payload()
        del missing['data']['records'][0]['credit_limit']
        cases.append(missing)
        extra = self.payload()
        extra['data']['records'][0]['secret'] = 'hidden'
        cases.append(extra)
        column_missing = self.payload()
        column_missing['data']['columns'].pop()
        cases.append(column_missing)
        reordered = self.payload()
        reordered['data']['columns'].reverse()
        cases.append(reordered)
        data_extra = self.payload()
        data_extra['data']['internal'] = True
        cases.append(data_extra)

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

    def test_malformed_monthly_summary_amounts_and_pagination_fail_closed(self):
        cases = []
        for summary in (
            'bad',
            [{'receivable_month': '2026-01'}],
            [{**self.payload()['data']['records'][0]['unverified_receivable_monthly_summary'][0], 'secret': 1}],
            [{**self.payload()['data']['records'][0]['unverified_receivable_monthly_summary'][0], 'receivable_month': []}],
            [{**self.payload()['data']['records'][0]['unverified_receivable_monthly_summary'][0], 'unverified_amount': True}],
            [{**self.payload()['data']['records'][0]['unverified_receivable_monthly_summary'][0], 'is_overdue': 1}],
        ):
            payload = self.payload()
            payload['data']['records'][0]['unverified_receivable_monthly_summary'] = summary
            cases.append(payload)

        bad_amount = self.payload()
        bad_amount['data']['records'][0]['credit_limit'] = '1000'
        cases.append(bad_amount)
        bad_name = self.payload()
        bad_name['data']['records'][0]['customer_name'] = []
        cases.append(bad_name)
        bad_record = self.payload()
        bad_record['data']['records'] = ['bad']
        cases.append(bad_record)
        bad_columns = self.payload()
        bad_columns['data']['columns'] = 'bad'
        cases.append(bad_columns)
        bad_column_item = self.payload()
        bad_column_item['data']['columns'][0]['internal'] = True
        cases.append(bad_column_item)
        bad_page = self.payload()
        bad_page['meta']['has_more'] = 1
        cases.append(bad_page)
        page_over_limit = self.payload()
        page_over_limit['meta']['page'] = 101
        cases.append(page_over_limit)
        extra_pagination = self.payload()
        extra_pagination['meta']['total'] = 1
        cases.append(extra_pagination)

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


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