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()