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_seventeen_tools(self): local = GatewayApp(api_client=RecordingApiClient()) public = PublicGatewayApp(None, None) self.assertEqual(local.registered_tool_names(), public.registered_tool_names()) self.assertEqual(17, 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()