mcp/tests/test_customer_payment_records_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_customer_payment_records']},
        }

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

    def test_closed_schema_requires_customer_and_forwards_dates(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_records', metadata['name'])
        self.assertEqual(['customer_id'], schema['required'])
        self.assertFalse(schema['additionalProperties'])
        self.assertEqual({
            'customer_id', 'receive_date_start', 'receive_date_end',
            'page', 'limit',
        }, set(schema['properties']))
        self.assertEqual('date', schema['properties']['receive_date_start']['format'])
        self.assertIn('list_customer_filter_options', metadata['description'])
        self.assertIn('不得根据名称猜测', metadata['description'])

        result = tool.call(
            customer_id=7,
            receive_date_start='2026-01-01',
            receive_date_end='2026-12-31',
            page=2,
            limit=30,
            request_id='rq_records',
        )
        self.assertEqual('MCP_0000', result['code'])
        self.assertEqual((
            'query_customer_payment_records',
            '/mcp/tools/queryCustomerPaymentRecords',
            {
                'customer_id': 7,
                'receive_date_start': '2026-01-01',
                'receive_date_end': '2026-12-31',
                'page': 2,
                'limit': 30,
            },
            'rq_records',
        ), client.calls[0])

    def test_invalid_arguments_and_oversized_date_range_fail(self):
        tool_class = self.load_tool_class()
        tool = tool_class(RecordingApiClient())
        invalid = (
            {'customer_id': True}, {'customer_id': 0}, {'customer_id': '7'},
            {'customer_id': 7, 'receive_date_start': '2026-02-30'},
            {'customer_id': 7, 'receive_date_start': 20260101},
            {'customer_id': 7, 'receive_date_start': '20260101'},
            {'customer_id': 7, 'receive_date_start': '2026-02-02', 'receive_date_end': '2026-02-01'},
            {'customer_id': 7, 'receive_date_start': '2025-01-01', 'receive_date_end': '2026-01-02'},
            {'customer_id': 7, 'page': 0}, {'customer_id': 7, 'page': 101},
            {'customer_id': 7, 'limit': False}, {'customer_id': 7, 'limit': 101},
        )
        for arguments in invalid:
            with self.subTest(arguments=arguments):
                with self.assertRaises(ValueError):
                    tool.call(**arguments)
        with self.assertRaises(RuntimeError):
            tool_class().call(customer_id=7)

        client = RecordingApiClient()
        tool = tool_class(client)
        tool.call(customer_id=7, receive_date_start='2026-01-01')
        self.assertEqual('2026-01-01', client.calls[-1][2]['receive_date_start'])
        self.assertNotIn('receive_date_end', client.calls[-1][2])
        tool.call(customer_id=7, receive_date_end='2026-12-31')
        self.assertEqual('2026-12-31', client.calls[-1][2]['receive_date_end'])
        self.assertNotIn('receive_date_start', client.calls[-1][2])
        tool.call(
            customer_id=7,
            receive_date_start='2025-01-01',
            receive_date_end='2026-01-01',
        )
        self.assertEqual('2026-01-01', client.calls[-1][2]['receive_date_end'])

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

        stdout = io.StringIO()
        local.run_cli([
            'call', '--tool', 'query_customer_payment_records',
            '--customer-id', '7', '--receive-date-start', '2026-01-01',
            '--receive-date-end', '2026-12-31', '--page', '2', '--limit', '30',
        ], stdout=stdout)
        self.assertEqual('MCP_0000', json.loads(stdout.getvalue())['code'])
        self.assertEqual({
            'customer_id': 7,
            'receive_date_start': '2026-01-01',
            'receive_date_end': '2026-12-31',
            'page': 2,
            'limit': 30,
        }, client.calls[-1][2])

        start_only = io.StringIO()
        local.run_cli([
            'call', '--tool', 'query_customer_payment_records',
            '--customer-id', '7', '--receive-date-start', '2026-01-01',
        ], stdout=start_only)
        self.assertIn('receive_date_start', client.calls[-1][2])
        self.assertNotIn('receive_date_end', client.calls[-1][2])

        end_only = io.StringIO()
        local.run_cli([
            'call', '--tool', 'query_customer_payment_records',
            '--customer-id', '7', '--receive-date-end', '2026-12-31',
        ], stdout=end_only)
        self.assertIn('receive_date_end', client.calls[-1][2])
        self.assertNotIn('receive_date_start', client.calls[-1][2])

        with self.assertRaises(ValueError):
            local.run_cli([
                'call', '--tool', 'query_customer_payment_records',
            ], stdout=io.StringIO())


class CustomerPaymentRecordsPresenterTest(unittest.TestCase):
    KEYS = [
        'customer_name', 'payment_reference', 'original_received_amount',
        'actual_received_amount', 'receive_date', 'verified_amount',
        'unverified_amount', 'payment_approval_status',
    ]
    HEADERS = [
        '客户名称', '收款水单号', '原币到账金额', '实际收款金额',
        '收款日期', '已核销金额', '未核销金额', '收款审核状态',
    ]

    def payload(self):
        return {
            'code': 'MCP_0000',
            'data': {
                'columns': [{'key': key, 'name': key} for key in self.KEYS],
                'records': [{
                    'customer_name': '甲客户',
                    'payment_reference': 'BANK-260001',
                    'original_received_amount': 100.25,
                    'actual_received_amount': 700.5,
                    'receive_date': '2026-07-20 16:30:00',
                    'verified_amount': 500.25,
                    'unverified_amount': 200.25,
                    'payment_approval_status': '审核通过',
                }],
            },
            'meta': {'page': 1, 'limit': 20, 'has_more': False},
        }

    def test_presenter_translates_exact_eight_columns(self):
        result = OutputPresenter().present(
            'query_customer_payment_records', self.payload()
        )
        self.assertFalse(result['is_error'])
        self.assertEqual(
            [{'label': header} for header in self.HEADERS],
            result['structured_content']['headers'],
        )
        self.assertEqual(8, 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_malformed_and_nonfinite_fields_fail_closed(self):
        cases = []
        unknown = self.payload()
        unknown['data']['records'][0]['secret'] = 'hidden'
        cases.append(unknown)
        missing = self.payload()
        del missing['data']['records'][0]['payment_reference']
        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)
        bad_records = self.payload()
        bad_records['data']['records'] = 'bad'
        cases.append(bad_records)
        bad_text = self.payload()
        bad_text['data']['records'][0]['customer_name'] = []
        cases.append(bad_text)
        bool_amount = self.payload()
        bool_amount['data']['records'][0]['verified_amount'] = True
        cases.append(bool_amount)
        for amount in (float('inf'), float('-inf'), float('nan'), '100.00'):
            bad_amount = self.payload()
            bad_amount['data']['records'][0]['actual_received_amount'] = amount
            cases.append(bad_amount)
        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)

        presenter = OutputPresenter()
        for payload in cases:
            with self.subTest(payload=payload):
                self.assertTrue(presenter.present(
                    'query_customer_payment_records', 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()