import copy import importlib import io import math from pathlib import Path 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_payable_cost_list', 'list_payable_cost_filter_options', 'export_payable_cost_list', ], }, } def call_tool(self, tool_code, route_path, payload, request_id): self.calls.append((tool_code, route_path, payload, request_id)) if tool_code == 'export_payable_cost_list': return { 'code': 'MCP_0000', 'data': { 'task_ref': 'mexp_payable', 'status': 'queued', 'retry_after_seconds': 10, }, } return {'code': 'MCP_0000', 'data': {}, 'meta': {}} class PayableCostToolContractTest(unittest.TestCase): FILTER_TYPES = ( '业务类型', '业务节点', '物流商', '费用项', '是否生成账单', '是否付款', '核销状态', '单据类型', ) QUERY_PROPERTIES = { 'business_type', 'order_numbers', 'tracking_numbers', 'container_codes', 'bl_numbers', 'so_numbers', 'business_date_start', 'business_date_end', 'cost_date_start', 'cost_date_end', 'operation_date_start', 'operation_date_end', 'business_node_id', 'provider_id', 'cost_type_ids', 'billing_status', 'payment_status', 'verification_status', 'document_type', 'page', 'limit', } def tool_class(self, module_name, class_name): path = Path(__file__).parents[1] / 'tools' / (module_name + '.py') self.assertTrue(path.is_file(), str(path)) return getattr(importlib.import_module('tools.' + module_name), class_name) def test_query_schema_is_closed_and_locks_required_business_type(self): cls = self.tool_class( 'query_payable_cost_list', 'QueryPayableCostListTool' ) metadata = cls().metadata() schema = metadata['input_schema'] self.assertEqual('query_payable_cost_list', metadata['name']) self.assertFalse(schema['additionalProperties']) self.assertEqual(self.QUERY_PROPERTIES, set(schema['properties'])) self.assertIn('business_type', schema['required']) self.assertEqual( [1, 2, 3, 4, 5, 7], schema['properties']['business_type']['enum'], ) self.assertIn('list_payable_cost_filter_options', metadata['description']) self.assertIn('禁止', metadata['description']) def test_query_call_forwards_five_number_fields_and_three_dates(self): cls = self.tool_class( 'query_payable_cost_list', 'QueryPayableCostListTool' ) client = RecordingApiClient() cls(client).call( business_type=1, order_numbers=[' O-1 '], tracking_numbers=[' T-1 '], container_codes=[' C-1 '], bl_numbers=[' B-1 '], so_numbers=[' S-1 '], business_date_start='2026-07-01', business_date_end='2026-07-31', cost_date_start='2026-07-02', cost_date_end='2026-07-30', operation_date_start='2026-07-03', operation_date_end='2026-07-29', business_node_id=11, provider_id=12, cost_type_ids=[13, 14], billing_status=0, payment_status=-1, verification_status=1, document_type=0, page=2, limit=100, request_id='rq_payable', ) self.assertEqual( ( 'query_payable_cost_list', '/mcp/tools/queryPayableCostList', { 'business_type': 1, 'order_numbers': ['O-1'], 'tracking_numbers': ['T-1'], 'container_codes': ['C-1'], 'bl_numbers': ['B-1'], 'so_numbers': ['S-1'], 'business_date_start': '2026-07-01', 'business_date_end': '2026-07-31', 'cost_date_start': '2026-07-02', 'cost_date_end': '2026-07-30', 'operation_date_start': '2026-07-03', 'operation_date_end': '2026-07-29', 'business_node_id': 11, 'provider_id': 12, 'cost_type_ids': [13, 14], 'billing_status': 0, 'payment_status': -1, 'verification_status': 1, 'document_type': 0, 'page': 2, 'limit': 100, }, 'rq_payable', ), client.calls[-1], ) def test_query_call_rejects_number_date_and_filter_boundaries(self): cls = self.tool_class( 'query_payable_cost_list', 'QueryPayableCostListTool' ) tool = cls(RecordingApiClient()) invalid = [ {}, {'business_type': 6, 'order_numbers': ['O']}, {'business_type': 1}, {'business_type': 1, 'order_numbers': 'O'}, {'business_type': 1, 'order_numbers': [1]}, {'business_type': 1, 'order_numbers': ['']}, {'business_type': 1, 'order_numbers': ['x' * 101]}, {'business_type': 1, 'order_numbers': [str(i) for i in range(201)]}, {'business_type': 1, 'order_numbers': ['O'] * 201}, {'business_type': 1, 'business_date_start': '2026-07-01'}, { 'business_type': 1, 'cost_date_start': '2026-07-01', 'cost_date_end': '2026-08-01', }, {'business_type': 1, 'operation_date_end': '2026-07-01'}, {'business_type': 1, 'order_numbers': ['O'], 'cost_type_ids': []}, {'business_type': 1, 'order_numbers': ['O'], 'cost_type_ids': [True]}, {'business_type': 1, 'order_numbers': ['O'], 'cost_type_ids': list(range(1, 202))}, {'business_type': 1, 'order_numbers': ['O'], 'page': 101}, {'business_type': 1, 'order_numbers': ['O'], 'limit': 0}, ] for arguments in invalid: with self.subTest(arguments=arguments): with self.assertRaises((TypeError, ValueError)): tool.call(**arguments) def test_filter_schema_and_call_lock_eight_types_and_cost_linkage(self): cls = self.tool_class( 'list_payable_cost_filter_options', 'ListPayableCostFilterOptionsTool', ) metadata = cls().metadata() schema = metadata['input_schema'] self.assertFalse(schema['additionalProperties']) self.assertEqual( {'filter_type', 'business_type', 'keyword', 'page', 'limit'}, set(schema['properties']), ) self.assertEqual( list(self.FILTER_TYPES), schema['properties']['filter_type']['enum'], ) client = RecordingApiClient() cls(client).call( filter_type=' 费用项 ', business_type=3, keyword=' 运 ', page=2, limit=30, ) self.assertEqual( { 'filter_type': '费用项', 'business_type': 3, 'keyword': '运', 'page': 2, 'limit': 30, }, client.calls[-1][2], ) for arguments in ( {'filter_type': '费用项'}, {'filter_type': '业务节点', 'business_type': 6}, {'filter_type': '未知'}, {'filter_type': '业务类型', 'keyword': 1}, {'filter_type': '业务类型', 'page': 0}, ): with self.subTest(arguments=arguments): with self.assertRaises((TypeError, ValueError)): cls(client).call(**arguments) def test_export_schema_matches_query_without_pagination(self): cls = self.tool_class( 'export_payable_cost_list', 'ExportPayableCostListTool' ) metadata = cls().metadata() schema = metadata['input_schema'] self.assertEqual('export_payable_cost_list', metadata['name']) self.assertFalse(schema['additionalProperties']) self.assertEqual( self.QUERY_PROPERTIES - {'page', 'limit'}, set(schema['properties']), ) self.assertIn('business_type', schema['required']) self.assertIn('query_export_task', metadata['description']) self.assertIn('单页签', metadata['description']) def test_local_public_registry_cli_and_safe_counts_are_current(self): local = GatewayApp().registered_tool_names() public = PublicGatewayApp(None, None).registered_tool_names() self.assertEqual(local, public) self.assertEqual(25, len(local)) self.assertEqual(24, len(OutputPresenter.SAFE_TOOLS)) self.assertEqual( ( 'query_payable_cost_list', 'list_payable_cost_filter_options', 'export_payable_cost_list', ), tuple(name for name in local if 'payable_cost' in name), ) client = RecordingApiClient() app = GatewayApp(api_client=client) app.run_cli([ 'call', '--tool', 'query_payable_cost_list', '--business-type', '1', '--order-numbers', ' O-1,O-2 ', '--cost-type-ids', '13,14', '--page', '2', '--limit', '30', ], stdout=io.StringIO()) self.assertEqual( { 'business_type': 1, 'order_numbers': ['O-1', 'O-2'], 'cost_type_ids': [13, 14], 'page': 2, 'limit': 30, }, client.calls[-1][2], ) app.run_cli([ 'call', '--tool', 'list_payable_cost_filter_options', '--filter-type', '费用项', '--business-type', '3', ], stdout=io.StringIO()) self.assertEqual('list_payable_cost_filter_options', client.calls[-1][0]) app.run_cli([ 'call', '--tool', 'export_payable_cost_list', '--business-type', '7', '--order-numbers', 'E-1', ], stdout=io.StringIO()) self.assertEqual('export_payable_cost_list', client.calls[-1][0]) def test_new_tool_error_paths_and_cli_forwarding_are_covered(self): query_cls = self.tool_class( 'query_payable_cost_list', 'QueryPayableCostListTool' ) with self.assertRaises(RuntimeError): query_cls().call(1, order_numbers=['O-1']) query = query_cls(RecordingApiClient()) query.call(business_type=1, order_numbers=['O', 'O']) invalid = ( { 'business_type': 1, 'business_date_start': '2026-07-02', 'business_date_end': '2026-07-01', }, { 'business_type': 1, 'business_date_start': 1, 'business_date_end': 1, }, { 'business_type': 1, 'business_date_start': '2026-99-01', 'business_date_end': '2026-99-02', }, { 'business_type': 1, 'business_date_start': '20260101', 'business_date_end': '20260102', }, {'business_type': 1, 'order_numbers': ['O'], 'billing_status': 2}, ) for arguments in invalid: with self.subTest(arguments=arguments): with self.assertRaises((TypeError, ValueError)): query.call(**arguments) export_cls = self.tool_class( 'export_payable_cost_list', 'ExportPayableCostListTool' ) with self.assertRaises(RuntimeError): export_cls().call(1, order_numbers=['O-1']) with self.assertRaises(ValueError): export_cls(RecordingApiClient()).call(6, order_numbers=['O-1']) filter_cls = self.tool_class( 'list_payable_cost_filter_options', 'ListPayableCostFilterOptionsTool', ) with self.assertRaises(RuntimeError): filter_cls().call('业务类型') filter_tool = filter_cls(RecordingApiClient()) with self.assertRaises(ValueError): filter_tool.call(1) with self.assertRaises(ValueError): filter_tool.call('业务类型', business_type=True) with self.assertRaises(ValueError): filter_tool.call('业务类型', keyword='x' * 101) filter_tool.call('业务类型') calls = [] app = GatewayApp() app.call_tool = lambda name, arguments, request_id='': ( calls.append((name, arguments, request_id)) or {'code': 'MCP_0000', 'data': {}, 'meta': {}} ) output = io.StringIO() app.run_cli([ 'call', '--tool', 'query_payable_cost_list', '--business-type', '1', '--order-numbers', 'O-1', '--tracking-numbers', 'T-1', '--container-codes', 'C-1', '--bl-numbers', 'B-1', '--so-numbers', 'S-1', '--business-date-start', '2026-07-01', '--business-date-end', '2026-07-02', '--cost-date-start', '2026-07-03', '--cost-date-end', '2026-07-04', '--operation-date-start', '2026-07-05', '--operation-date-end', '2026-07-06', '--business-node-id', '11', '--provider-id', '12', '--cost-type-ids', '13,14', '--billing-status', '1', '--payment-status', '-1', '--verification-status', '2', '--document-type', '0', ], stdout=output) self.assertEqual('query_payable_cost_list', calls[-1][0]) app.run_cli([ 'call', '--tool', 'export_payable_cost_list', '--business-type', '1', '--order-numbers', 'O-1', ], stdout=io.StringIO()) self.assertEqual('export_payable_cost_list', calls[-1][0]) with self.assertRaises(ValueError): app.run_cli(['call', '--tool', 'query_payable_cost_list'], stdout=io.StringIO()) app.run_cli([ 'call', '--tool', 'list_payable_cost_filter_options', '--filter-type', '业务类型', ], stdout=io.StringIO()) app.run_cli([ 'call', '--tool', 'list_payable_cost_filter_options', '--filter-type', '业务类型', '--business-type', '3', ], stdout=io.StringIO()) with self.assertRaises(ValueError): app.run_cli([ 'call', '--tool', 'list_payable_cost_filter_options' ], stdout=io.StringIO()) for argv in ( [ 'call', '--tool', 'query_customer_payment_followup', '--customer-id', '1', '--department-id', '2', '--sales-id', '3', '--merchandiser-id', '4', '--has-unverified-receivable-only', 'false', ], [ 'call', '--tool', 'query_customer_unverified_bill_details', '--customer-id', '1', ], [ 'call', '--tool', 'query_customer_payment_records', '--customer-id', '1', '--receive-date-start', '2026-07-01', '--receive-date-end', '2026-07-02', ], [ 'call', '--tool', 'query_order_receivable_cost_details', '--order-number', 'O-1', ], ): app.run_cli(argv, stdout=io.StringIO()) app.run_cli([ 'call', '--tool', 'query_customer_payment_followup', ], stdout=io.StringIO()) for argv in ( ['call', '--tool', 'query_customer_unverified_bill_details'], ['call', '--tool', 'query_customer_payment_records'], ['call', '--tool', 'query_order_receivable_cost_details'], ): with self.assertRaises(ValueError): app.run_cli(argv, stdout=io.StringIO()) app.run_cli([ 'call', '--tool', 'query_customer_payment_records', '--customer-id', '1', '--receive-date-start', '2026-07-01', ], stdout=io.StringIO()) app.run_cli([ 'call', '--tool', 'query_customer_payment_records', '--customer-id', '1', '--receive-date-end', '2026-07-02', ], stdout=io.StringIO()) class PayableCostPresenterContractTest(unittest.TestCase): COLUMN_MAP = { 1: [ ('number', '单号'), ('so_number', 'SO号'), ('container_code', '柜号'), ('bl_number', '提单号'), ('sub_number', '订单号'), ('business_node_name', '业务节点'), ('providers_name', '物流商'), ('cost_name', '费用名称'), ('payable_cost_q', '金额'), ('currency', '币种'), ('company_money_q', '本位币金额'), ('yf_lock', '应付锁定'), ('trade_date', '费用发生时间'), ('verify_status_text', '核销状态'), ('remark', '备注'), ('bill_no', '账单编号'), ('payment_order_number', '付款单号'), ], 2: [ ('number', '单号'), ('sub_number', '跟踪单号'), ('workorder_number', '工单号'), ('order_type_name', '单据类型'), ('providers_name', '物流商'), ('cost_name', '费用名称'), ('payable_cost_q', '应付原币金额'), ('currency', '应付原币币种'), ('company_money_q', '本位币金额CNY'), ('yf_lock', '应付锁定'), ('business_date', '业务发生时间'), ('trade_date', '费用发生时间'), ('create_date', '操作时间'), ('verify_status_text', '核销状态'), ('remark', '备注'), ('bill_no', '账单编号'), ('payment_order_number', '付款单号'), ], 3: [ ('number', '单号'), ('order_status_text', '包裹状态'), ('providers_name', '物流商'), ('business_node_name', '费用业务节点'), ('cost_name', '费用名称'), ('detail_price', '计费单价'), ('charge_weight', '计费重'), ('quote_cost', '计费金额'), ('payable_cost_q', '实际应付金额'), ('quote_currency', '原币币种'), ('company_money_q', '计费本位币金额'), ('yf_lock', '应付锁定'), ('bill_no', '账单编号'), ('payment_order_number', '付款单号'), ('trade_date', '费用发生时间'), ('verify_status_text', '核销状态'), ('remark', '财务备注'), ('create_user_name', '操作人'), ('create_date', '录入时间'), ], 4: [ ('number', '订单号'), ('business_bill_no', '货代账单号'), ('providers_name', '物流商'), ('cost_name', '费用名称'), ('payable_cost_q', '金额'), ('currency', '币种'), ('company_money_q', '本位币金额'), ('department_name', '事业部'), ('yf_lock', '应付锁定'), ('trade_date', '费用发生时间'), ('verify_status_text', '核销状态'), ('business_node_name', '业务节点'), ('remark', '备注'), ('bill_no', '应付账单编号'), ('payment_order_number', '付款单号'), ], 5: [ ('order_type_name', '单据类型'), ('number', '订单号'), ('warehouse_name', '仓库'), ('business_node_name', '业务节点'), ('providers_name', '物流商'), ('cost_name', '费用名称'), ('payable_cost_q', '金额'), ('currency', '币种'), ('company_money_q', '本位币金额'), ('yf_lock', '应付锁定'), ('trade_date', '费用发生时间'), ('verify_status_text', '核销状态'), ('remark', '备注'), ('bill_no', '账单编号'), ('payment_order_number', '付款单号'), ], 7: [ ('number', '订单号'), ('ep_order_status_text', '订单状态'), ('business_node_name', '业务节点'), ('providers_name', '物流商'), ('cost_name', '费用名称'), ('detail_price', '计费单价'), ('charge_weight', '计费重'), ('quote_cost', '计费金额'), ('payable_cost_q', '实际应付金额'), ('quote_currency', '原币币种'), ('company_money_q', '本位币金额'), ('trade_date', '费用发生时间'), ('remark', '备注'), ('verify_status_text', '核销状态'), ('bill_no', '账单编号'), ('payment_order_number', '付款单号'), ('create_user_name', '操作人'), ('create_date', '录入时间'), ], } AMOUNT_KEYS = { 'detail_price', 'charge_weight', 'quote_cost', 'payable_cost_q', 'company_money_q', } def payload(self, business_type): columns = self.COLUMN_MAP[business_type] record = {} for key, _ in columns: record[key] = 12.5 if key in self.AMOUNT_KEYS else key + '-value' return { 'code': 'MCP_0000', 'data': { 'business_type': business_type, 'company_currency': 'CNY', 'columns': [ {'key': key, 'name': name} for key, name in columns ], 'records': [record], }, 'meta': { 'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_payable', }, } def test_six_business_types_have_exact_dynamic_columns(self): presenter = OutputPresenter() for business_type, columns in self.COLUMN_MAP.items(): with self.subTest(business_type=business_type): result = presenter.present( 'query_payable_cost_list', self.payload(business_type) ) self.assertFalse(result['is_error']) self.assertEqual( [{'label': name} for _, name in columns], result['structured_content']['headers'], ) self.assertEqual( len(columns), len(result['structured_content']['rows'][0]), ) self.assertEqual( business_type, result['structured_content']['business_type'], ) def test_unknown_missing_wrong_order_and_non_finite_values_fail_closed(self): presenter = OutputPresenter() self.assertTrue(presenter.handles('query_payable_cost_list')) cases = [] payload = self.payload(1) payload['data']['columns'].append({'key': 'secret', 'name': '秘密'}) payload['data']['records'][0]['secret'] = 'hidden' cases.append(payload) payload = self.payload(1) payload['data']['columns'].reverse() cases.append(payload) payload = self.payload(1) payload['data']['records'][0].pop('cost_name') cases.append(payload) payload = self.payload(1) payload['data']['records'][0]['secret'] = 'hidden' cases.append(payload) payload = self.payload(6) if 6 in self.COLUMN_MAP else self.payload(1) payload['data']['business_type'] = 6 cases.append(payload) for business_type in self.COLUMN_MAP: for key, _ in self.COLUMN_MAP[business_type]: if key not in self.AMOUNT_KEYS: continue for value in (True, '12.5', math.inf, -math.inf, math.nan): payload = self.payload(business_type) payload['data']['records'][0][key] = value cases.append(payload) for payload in cases: with self.subTest(payload=payload): self.assertTrue(presenter.present( 'query_payable_cost_list', payload )['is_error']) def test_filter_and_export_presenters_use_safe_contracts(self): presenter = OutputPresenter() filtered = presenter.present( 'list_payable_cost_filter_options', { 'code': 'MCP_0000', 'data': { 'records': [ {'value': 0, 'label': '否', 'code': 'no'}, {'value': -1, 'label': '未付款', 'code': 'unpaid'}, ], }, 'meta': {'page': 1, 'limit': 20, 'has_more': False}, }, ) self.assertFalse(filtered['is_error']) self.assertEqual( [[0, '否', 'no'], [-1, '未付款', 'unpaid']], filtered['structured_content']['rows'], ) exported = presenter.present( 'export_payable_cost_list', { 'code': 'MCP_0000', 'data': { 'task_ref': 'mexp_payable', 'status': 'queued', 'retry_after_seconds': 10, }, }, ) self.assertFalse(exported['is_error']) self.assertEqual( 'mexp_payable', exported['structured_content']['task']['task_ref'] ) def test_payable_filter_presenter_rejects_malformed_records_and_meta(self): presenter = OutputPresenter() valid = { 'code': 'MCP_0000', 'data': { 'records': [{'value': 0, 'label': 'ok', 'code': 'no'}], }, 'meta': {'page': 1, 'limit': 20, 'has_more': False}, } cases = [] malformed = copy.deepcopy(valid) malformed['data']['unexpected'] = True cases.append(malformed) malformed = copy.deepcopy(valid) malformed['data']['records'][0].pop('code') cases.append(malformed) malformed = copy.deepcopy(valid) malformed['data']['records'][0]['value'] = '0' cases.append(malformed) malformed = copy.deepcopy(valid) malformed['data']['records'][0]['label'] = '' cases.append(malformed) malformed = copy.deepcopy(valid) malformed['meta'].pop('has_more') cases.append(malformed) malformed = copy.deepcopy(valid) malformed['meta']['unexpected'] = True cases.append(malformed) for payload in cases: with self.subTest(payload=payload): self.assertTrue( presenter.present( 'list_payable_cost_filter_options', payload )['is_error'] ) def test_presenter_rejects_unknown_top_level_and_non_string_text_cells(self): presenter = OutputPresenter() payload = self.payload(1) payload['data']['unexpected'] = True self.assertTrue( presenter.present('query_payable_cost_list', payload)['is_error'] ) payload = self.payload(1) payload['data']['records'][0]['remark'] = 7 self.assertTrue( presenter.present('query_payable_cost_list', payload)['is_error'] ) if __name__ == '__main__': unittest.main()