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_order_receivable_cost_details']}, } 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 OrderReceivableCostDetailsToolTest(unittest.TestCase): def load_tool_class(self): path = os.path.join( os.path.dirname(os.path.dirname(__file__)), 'tools', 'query_order_receivable_cost_details.py', ) self.assertTrue(os.path.isfile(path), '订单应收费用 Gateway 工具尚未实现') spec = importlib.util.spec_from_file_location('receivable_cost_tool', path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module.QueryOrderReceivableCostDetailsTool def test_closed_schema_requires_explicit_order_number_and_forwards_exactly(self): tool_class = self.load_tool_class() client = RecordingApiClient() metadata = tool_class(client).metadata() schema = metadata['input_schema'] self.assertEqual('query_order_receivable_cost_details', metadata['name']) self.assertEqual(['order_number'], schema['required']) self.assertFalse(schema['additionalProperties']) self.assertEqual({'order_number', 'page', 'limit'}, set(schema['properties'])) self.assertIn('明确订单号', metadata['description']) self.assertIn('不得使用参考号', metadata['description']) self.assertIn('不得模糊', metadata['description']) result = tool_class(client).call( order_number='ORD-260001', page=2, limit=30, request_id='rq_cost' ) self.assertEqual('MCP_0000', result['code']) self.assertEqual(( 'query_order_receivable_cost_details', '/mcp/tools/queryOrderReceivableCostDetails', {'order_number': 'ORD-260001', 'page': 2, 'limit': 30}, 'rq_cost', ), client.calls[-1]) def test_invalid_arguments_fail_before_backend_call(self): tool_class = self.load_tool_class() client = RecordingApiClient() tool = tool_class(client) for arguments in ( {}, {'order_number': ''}, {'order_number': ' '}, {'order_number': 123}, {'order_number': True}, {'order_number': 'x' * 101}, {'order_number': 'ORD-1', 'page': 0}, {'order_number': 'ORD-1', 'page': 101}, {'order_number': 'ORD-1', 'limit': False}, {'order_number': 'ORD-1', 'limit': 101}, ): with self.subTest(arguments=arguments): with self.assertRaises((TypeError, ValueError)): tool.call(**arguments) self.assertEqual([], client.calls) with self.assertRaises(RuntimeError): tool_class().call(order_number='ORD-1') def test_registries_cli_and_payment_records_remain_available(self): client = RecordingApiClient() local = GatewayApp(api_client=client) public = PublicGatewayApp(None, None) self.assertEqual(local.registered_tool_names(), public.registered_tool_names()) self.assertEqual(19, len(local.registered_tool_names())) self.assertIn('query_order_receivable_cost_details', local.registered_tool_names()) self.assertIn('query_customer_payment_records', local.registered_tool_names()) stdout = io.StringIO() local.run_cli([ 'call', '--tool', 'query_order_receivable_cost_details', '--order-number', 'ORD-260001', '--page', '2', '--limit', '30', ], stdout=stdout) self.assertEqual('MCP_0000', json.loads(stdout.getvalue())['code']) self.assertEqual('ORD-260001', client.calls[-1][2]['order_number']) with self.assertRaises(ValueError): local.run_cli([ 'call', '--tool', 'query_order_receivable_cost_details', ], stdout=io.StringIO()) class OrderReceivableCostDetailsPresenterTest(unittest.TestCase): KEYS = [ 'order_number', 'customer_name', 'cost_name', 'cost_customer_name', 'charge_weight', 'unit_price', 'quote_currency', 'receivable_original_amount', 'receivable_amount_cny', 'settlement_amount_cny', 'cost_confirmation_status', 'closing_status', 'verification_status', ] HEADERS = [ '订单号', '客户名称', '费用项', '费用客户', '计费重', '计费单价', '原币币种', '应收原币', '应收金额（CNY）', '结算金额（CNY）', '费用确认状态', '关账状态', '核销状态', ] def payload(self): return { 'code': 'MCP_0000', 'data': { 'columns': [{'key': key, 'name': key} for key in self.KEYS], 'records': [{ 'order_number': 'ORD-260001', 'customer_name': '主客户', 'cost_name': '海运费', 'cost_customer_name': '子客户（SC01）', 'charge_weight': 12.35, 'unit_price': 8.5, 'quote_currency': 'USD', 'receivable_original_amount': 10.0, 'receivable_amount_cny': 72.0, 'settlement_amount_cny': 39.0, 'cost_confirmation_status': '已确认', 'closing_status': '部分关账', 'verification_status': '是', }], }, 'meta': {'page': 1, 'limit': 20, 'has_more': False}, } def test_presenter_translates_exact_thirteen_columns(self): result = OutputPresenter().present( 'query_order_receivable_cost_details', self.payload() ) self.assertFalse(result['is_error']) self.assertEqual( [{'label': header} for header in self.HEADERS], result['structured_content']['headers'], ) self.assertEqual(13, 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_reordered_and_bad_values_fail_closed(self): cases = [] unknown = self.payload() unknown['data']['records'][0]['secret'] = 'hidden' cases.append(unknown) missing = self.payload() del missing['data']['records'][0]['cost_name'] 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) 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) bad_text = self.payload() bad_text['data']['records'][0]['order_number'] = [] cases.append(bad_text) for amount in (True, '10.00', float('inf'), float('-inf'), float('nan')): bad_amount = self.payload() bad_amount['data']['records'][0]['receivable_amount_cny'] = amount cases.append(bad_amount) presenter = OutputPresenter() for payload in cases: with self.subTest(payload=payload): self.assertTrue(presenter.present( 'query_order_receivable_cost_details', 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()