import copy import importlib.util import io import json import math import os import unittest from app import GatewayApp from public_gateway import PublicGatewayApp from services.output_presenter import OutputPresenter ROOT = os.path.dirname(os.path.dirname(__file__)) class RecordingApiClient: def __init__(self): self.calls = [] def list_enabled_tools(self, request_id=''): return { 'code': 'MCP_0000', 'data': { 'tool_codes': [ 'query_receivable_cost_list', 'list_receivable_cost_filter_options', ], }, } 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 ToolLoaderMixin: def load_tool(self, filename, class_name): path = os.path.join(ROOT, 'tools', filename) self.assertTrue(os.path.isfile(path), filename + ' is not implemented') spec = importlib.util.spec_from_file_location(class_name, path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return getattr(module, class_name) class ReceivableCostListToolTest(ToolLoaderMixin, unittest.TestCase): NUMBER_FIELDS = ( 'reference_numbers', 'tracking_numbers', 'order_numbers', 'bill_numbers', ) def tool_class(self): return self.load_tool( 'query_receivable_cost_list.py', 'QueryReceivableCostListTool', ) def test_schema_is_closed_exact_and_zero_inference(self): metadata = self.tool_class()().metadata() schema = metadata['input_schema'] self.assertEqual('query_receivable_cost_list', metadata['name']) self.assertFalse(schema['additionalProperties']) self.assertNotIn('number', schema['properties']) self.assertEqual({ 'reference_numbers', 'tracking_numbers', 'order_numbers', 'bill_numbers', 'business_date_start', 'business_date_end', 'customer_id', 'sub_customer_id', 'billing_status', 'verification_status', 'document_type', 'cost_type_id', 'page', 'limit', }, set(schema['properties'])) for field in self.NUMBER_FIELDS: definition = schema['properties'][field] self.assertEqual('array', definition['type']) self.assertEqual(1, definition['minItems']) self.assertEqual(200, definition['maxItems']) self.assertEqual({ 'type': 'string', 'minLength': 1, 'maxLength': 100, 'pattern': '.*\\S.*', }, definition['items']) self.assertEqual([0, 1], schema['properties']['billing_status']['enum']) self.assertEqual( [-1, 0, 1], schema['properties']['verification_status']['enum'], ) self.assertEqual(0, schema['properties']['document_type']['minimum']) self.assertEqual(1, schema['properties']['cost_type_id']['minimum']) self.assertIn('禁止根据格式猜测', metadata['description']) self.assertIn( 'list_receivable_cost_filter_options', metadata['description'], ) def test_call_forwards_exact_filters_and_date_pair(self): tool_class = self.tool_class() client = RecordingApiClient() result = tool_class(client).call( reference_numbers=[' REF-1 '], tracking_numbers=['TRACK-1'], order_numbers=['ORDER-1'], bill_numbers=['BILL-1'], business_date_start='2026-07-01', business_date_end='2026-07-31', customer_id=7, sub_customer_id=9, billing_status=0, verification_status=-1, document_type=0, cost_type_id=12, page=2, limit=30, request_id='rq_receivable', ) self.assertEqual('MCP_0000', result['code']) self.assertEqual(( 'query_receivable_cost_list', '/mcp/tools/queryReceivableCostList', { 'reference_numbers': ['REF-1'], 'tracking_numbers': ['TRACK-1'], 'order_numbers': ['ORDER-1'], 'bill_numbers': ['BILL-1'], 'business_date_start': '2026-07-01', 'business_date_end': '2026-07-31', 'customer_id': 7, 'sub_customer_id': 9, 'billing_status': 0, 'verification_status': -1, 'document_type': 0, 'cost_type_id': 12, 'page': 2, 'limit': 30, }, 'rq_receivable', ), client.calls[-1]) def test_call_enforces_cross_field_and_strict_boundaries(self): tool = self.tool_class()(RecordingApiClient()) invalid = [ {}, {'business_date_start': '2026-07-01'}, {'business_date_end': '2026-07-31'}, { 'business_date_start': '2026-07-01', 'business_date_end': '2026-08-01', }, {'reference_numbers': []}, {'reference_numbers': [7]}, {'reference_numbers': [' ']}, {'reference_numbers': ['x' * 101]}, {'reference_numbers': ['x'] * 201}, { 'reference_numbers': ['x'] * 101, 'tracking_numbers': ['y'] * 100, }, {'reference_numbers': ['x'], 'sub_customer_id': 9}, {'reference_numbers': ['x'], 'customer_id': True}, {'reference_numbers': ['x'], 'billing_status': 2}, {'reference_numbers': ['x'], 'verification_status': 2}, {'reference_numbers': ['x'], 'document_type': -1}, {'reference_numbers': ['x'], 'cost_type_id': 0}, {'reference_numbers': ['x'], 'page': 101}, {'reference_numbers': ['x'], 'limit': 1.5}, { 'reference_numbers': ['x'], 'business_date_start': 20260701, 'business_date_end': '2026-07-31', }, { 'reference_numbers': ['x'], 'business_date_start': 'not-a-date', 'business_date_end': '2026-07-31', }, { 'reference_numbers': ['x'], 'business_date_start': '20260701', 'business_date_end': '2026-07-31', }, ] for arguments in invalid: with self.subTest(arguments=arguments): with self.assertRaises((TypeError, ValueError)): tool.call(**arguments) self.assertEqual([], tool.api_client.calls) with self.assertRaisesRegex(RuntimeError, 'api client is required'): self.tool_class()().call(reference_numbers=['REF-1']) class ReceivableCostFilterOptionsToolTest(ToolLoaderMixin, unittest.TestCase): FILTER_TYPES = ['主客户', '子客户', '出账状态', '核销状态', '单据类型', '费用项'] def tool_class(self): return self.load_tool( 'list_receivable_cost_filter_options.py', 'ListReceivableCostFilterOptionsTool', ) def test_schema_has_exact_types_and_customer_dependency(self): metadata = self.tool_class()().metadata() schema = metadata['input_schema'] self.assertFalse(schema['additionalProperties']) self.assertEqual(['filter_type'], schema['required']) self.assertEqual( self.FILTER_TYPES, schema['properties']['filter_type']['enum'], ) self.assertEqual( {'type': 'integer', 'minimum': 1}, schema['properties']['customer_id'], ) self.assertIn('子客户', json.dumps(schema, ensure_ascii=False)) self.assertIn('禁止猜测', metadata['description']) def test_call_forwards_and_enforces_customer_linkage(self): tool_class = self.tool_class() client = RecordingApiClient() tool = tool_class(client) tool.call( filter_type=' 子客户 ', customer_id=7, keyword=' 子 ', page=2, limit=30, request_id='rq_filters', ) self.assertEqual(( 'list_receivable_cost_filter_options', '/mcp/tools/listReceivableCostFilterOptions', { 'filter_type': '子客户', 'customer_id': 7, 'keyword': '子', 'page': 2, 'limit': 30, }, 'rq_filters', ), client.calls[-1]) tool.call(filter_type='主客户') for arguments in ( {'filter_type': '子客户'}, {'filter_type': 7}, {'filter_type': '子客户', 'customer_id': 0}, {'filter_type': '主客户', 'customer_id': 7}, {'filter_type': '其他'}, {'filter_type': '主客户', 'keyword': 7}, {'filter_type': '主客户', 'keyword': 'x' * 101}, {'filter_type': '主客户', 'page': 0}, {'filter_type': '主客户', 'limit': True}, {'filter_type': '主客户', 'limit': 101}, ): with self.subTest(arguments=arguments): with self.assertRaises((TypeError, ValueError)): tool.call(**arguments) self.assertEqual(2, len(client.calls)) with self.assertRaisesRegex(RuntimeError, 'api client is required'): tool_class().call('主客户') class ReceivableCostIntegrationTest(unittest.TestCase): def test_registries_are_identical_ordered_and_counts_are_current(self): local = GatewayApp().registered_tool_names() public = PublicGatewayApp(None, None).registered_tool_names() self.assertEqual(local, public) self.assertEqual(22, len(local)) self.assertEqual(21, len(OutputPresenter.SAFE_TOOLS)) self.assertEqual( ( 'query_receivable_cost_list', 'list_receivable_cost_filter_options', ), tuple( name for name in local if name in { 'query_receivable_cost_list', 'list_receivable_cost_filter_options', } ), ) def test_cli_forwards_both_tools(self): client = RecordingApiClient() app = GatewayApp(api_client=client) app.run_cli([ 'call', '--tool', 'query_receivable_cost_list', '--reference-numbers', ' REF-1,REF-2 ', '--bill-numbers', ' BILL-1 ', '--business-date-start', '2026-07-01', '--business-date-end', '2026-07-31', '--customer-id', '7', '--sub-customer-id', '9', '--billing-status', '0', '--verification-status', '-1', '--document-type', '0', '--cost-type-id', '12', '--page', '2', '--limit', '30', ], stdout=io.StringIO()) self.assertEqual({ 'reference_numbers': ['REF-1', 'REF-2'], 'bill_numbers': ['BILL-1'], 'business_date_start': '2026-07-01', 'business_date_end': '2026-07-31', 'customer_id': 7, 'sub_customer_id': 9, 'billing_status': 0, 'verification_status': -1, 'document_type': 0, 'cost_type_id': 12, 'page': 2, 'limit': 30, }, client.calls[-1][2]) app.run_cli([ 'call', '--tool', 'query_receivable_cost_list', '--order-numbers', 'ORDER-1', ], stdout=io.StringIO()) self.assertEqual({ 'order_numbers': ['ORDER-1'], 'page': 1, 'limit': 20, }, client.calls[-1][2]) app.run_cli([ 'call', '--tool', 'list_receivable_cost_filter_options', '--filter-type', '子客户', '--customer-id', '7', '--keyword', '子', ], stdout=io.StringIO()) self.assertEqual({ 'filter_type': '子客户', 'customer_id': 7, 'keyword': '子', 'page': 1, 'limit': 20, }, client.calls[-1][2]) app.run_cli([ 'call', '--tool', 'list_receivable_cost_filter_options', '--filter-type', '主客户', ], stdout=io.StringIO()) self.assertEqual({ 'filter_type': '主客户', 'keyword': '', 'page': 1, 'limit': 20, }, client.calls[-1][2]) with self.assertRaisesRegex(ValueError, '--filter-type is required'): app.run_cli([ 'call', '--tool', 'list_receivable_cost_filter_options', ], stdout=io.StringIO()) class ReceivableCostPresenterTest(unittest.TestCase): KEYS = [ 'main_customer_name', 'sub_customer_name', 'customer_attribute', 'department_name', 'document_type', 'warehouse_name', 'business_number', 'reference_number', 'cost_name', 'original_amount', 'total_amount', 'billed_amount', 'unbilled_amount', 'bill_number', 'verification_status', 'receipt_number', 'receipt_date', 'sales_name', 'merchandiser_name', 'drainage_user_name', 'first_leg_order_status', 'settlement_mode', 'cost_occurred_at', 'business_occurred_at', ] NAMES = [ '主客户', '子客户', '客户属性', '事业部', '单据类型', '仓库', '业务单号', '参考号', '费用项', '原币金额', '总金额', '已出账金额', '未出账金额', '账单编号', '核销状态', '收款水单号', '收款日期', '商务经理', '客户经理', '引流人', '头程订单状态', '结算模式', '费用发生时间', '业务发生时间', ] AMOUNT_KEYS = { 'original_amount', 'total_amount', 'billed_amount', 'unbilled_amount', } def payload(self): record = {key: '值' for key in self.KEYS} record.update({ 'original_amount': {'amount': 10.25, 'currency': 'USD'}, 'total_amount': {'amount': 70.5, 'currency': 'CNY'}, 'billed_amount': {'amount': 20.0, 'currency': 'CNY'}, 'unbilled_amount': {'amount': 50.5, 'currency': 'CNY'}, }) return { 'code': 'MCP_0000', 'data': { 'columns': [ {'key': key, 'name': name} for key, name in zip(self.KEYS, self.NAMES) ], 'records': [record], }, 'meta': { 'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_receivable', }, } def test_query_presenter_requires_exact_contract(self): result = OutputPresenter().present( 'query_receivable_cost_list', self.payload(), ) self.assertFalse(result['is_error']) self.assertEqual( [{'label': name} for name in self.NAMES], result['structured_content']['headers'], ) self.assertEqual(24, len(result['structured_content']['rows'][0])) self.assertEqual( {'page': 1, 'limit': 20, 'has_more': False}, result['structured_content']['pagination'], ) def test_query_presenter_fails_closed_on_malformed_contract(self): cases = [] for mutation in ( lambda value: value['data']['columns'].reverse(), lambda value: value['data']['columns'][0].update({'secret': True}), lambda value: value['data']['columns'][0].update({'name': '错误名称'}), lambda value: value['data']['records'][0].update({'secret': 'hidden'}), lambda value: value['data']['records'][0].pop('cost_name'), lambda value: value['data']['records'][0].update({'cost_name': []}), lambda value: value['data'].update({'secret': True}), lambda value: value['meta'].update({'total': 1}), ): payload = self.payload() mutation(payload) cases.append(payload) for key in self.AMOUNT_KEYS: for amount in (True, '10', float('inf'), float('-inf'), float('nan')): payload = self.payload() payload['data']['records'][0][key]['amount'] = amount cases.append(payload) for malformed in ( {'amount': 1}, {'amount': 1, 'currency': 'CNY', 'secret': True}, {'amount': 1, 'currency': 7}, ): payload = self.payload() payload['data']['records'][0][key] = malformed cases.append(payload) for meta in ( {'page': 0, 'limit': 20, 'has_more': False}, {'page': 1, 'limit': 101, 'has_more': False}, {'page': 1, 'limit': 20, 'has_more': 0}, ): payload = self.payload() payload['meta'] = meta cases.append(payload) presenter = OutputPresenter() for payload in cases: with self.subTest(payload=payload): self.assertTrue(presenter.present( 'query_receivable_cost_list', payload )['is_error']) def test_filter_presenter_preserves_zero_and_negative_values(self): payload = { 'code': 'MCP_0000', 'data': { 'records': [ {'value': 0, 'label': '未出账', 'code': 'unbilled'}, {'value': -1, 'label': '未核销', 'code': 'unverified'}, ], }, 'meta': {'page': 1, 'limit': 20, 'has_more': False}, } result = OutputPresenter().present( 'list_receivable_cost_filter_options', payload, ) self.assertFalse(result['is_error']) self.assertEqual( [[0, '未出账', 'unbilled'], [-1, '未核销', 'unverified']], result['structured_content']['rows'], ) for mutation in ( lambda value: value['data'].update({'secret': True}), lambda value: value['data']['records'][0].update({'secret': True}), lambda value: value['data']['records'][0].update({'value': True}), lambda value: value['data']['records'][0].update({'label': ''}), lambda value: value['data']['records'][0].update({'code': 1}), lambda value: value['meta'].update({'total': 2}), ): malformed = copy.deepcopy(payload) mutation(malformed) self.assertTrue(OutputPresenter().present( 'list_receivable_cost_filter_options', malformed, )['is_error']) if __name__ == '__main__': unittest.main()