import io import unittest from app import GatewayApp from public_gateway import PublicGatewayApp from services.output_presenter import OutputPresenter from tools.export_receivable_cost_list import ExportReceivableCostListTool class RecordingApiClient: def __init__(self): self.calls = [] def list_enabled_tools(self, request_id=''): return { 'code': 'MCP_0000', 'data': {'tool_codes': ['export_receivable_cost_list']}, } 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': { 'task_ref': 'mexp_test', 'status': 'queued', 'retry_after_seconds': 10, }, } class ExportReceivableCostListToolTest(unittest.TestCase): def test_schema_is_closed_and_matches_list_filters_without_pagination(self): metadata = ExportReceivableCostListTool().metadata() schema = metadata['input_schema'] self.assertEqual('export_receivable_cost_list', metadata['name']) self.assertFalse(schema['additionalProperties']) 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', }, set(schema['properties'])) self.assertNotIn('page', schema['properties']) self.assertNotIn('limit', schema['properties']) self.assertNotIn('ids', schema['properties']) self.assertIn('明确要求导出应收费用单', metadata['description']) self.assertIn('query_receivable_cost_list', metadata['description']) self.assertIn('query_export_task', metadata['description']) self.assertIn('禁止在一次调用内轮询', metadata['description']) def test_call_forwards_exact_filters(self): client = RecordingApiClient() result = ExportReceivableCostListTool(client).call( order_numbers=[' ORDER-1 '], billing_status=0, ) self.assertEqual('MCP_0000', result['code']) self.assertEqual( ( 'export_receivable_cost_list', '/mcp/tools/exportReceivableCostList', { 'order_numbers': ['ORDER-1'], 'billing_status': 0, }, 'rq_export_receivable_cost_list', ), client.calls[0], ) def test_call_validation_boundaries(self): tool = ExportReceivableCostListTool() with self.assertRaisesRegex(RuntimeError, 'api client is required'): tool.call(order_numbers=['O1']) client = RecordingApiClient() tool = ExportReceivableCostListTool(client) with self.assertRaisesRegex(ValueError, 'non-empty list'): tool.call(order_numbers='O1') with self.assertRaisesRegex(ValueError, 'must be strings'): tool.call(order_numbers=[1]) with self.assertRaisesRegex(ValueError, '1 to 100 chars'): tool.call(order_numbers=['']) with self.assertRaisesRegex(ValueError, '1 to 100 chars'): tool.call(order_numbers=['x' * 101]) with self.assertRaisesRegex(ValueError, 'at most 200'): tool.call(order_numbers=['n{0}'.format(i) for i in range(201)]) with self.assertRaisesRegex(ValueError, 'both start and end'): tool.call(business_date_start='2026-07-01') with self.assertRaisesRegex(ValueError, 'within 31 days'): tool.call( business_date_start='2026-07-01', business_date_end='2026-08-01', ) with self.assertRaisesRegex(ValueError, 'number filters or a business'): tool.call() with self.assertRaisesRegex(ValueError, 'requires customer_id'): tool.call(order_numbers=['O1'], sub_customer_id=9) ExportReceivableCostListTool(client).call( order_numbers=['ORDER-1', ' ORDER-1 '], ) self.assertEqual({'order_numbers': ['ORDER-1']}, client.calls[-1][2]) ExportReceivableCostListTool(client).call( business_date_start='2026-07-01', business_date_end='2026-07-31', customer_id=7, sub_customer_id=9, verification_status=-1, document_type=8, cost_type_id=12, ) self.assertEqual( { 'business_date_start': '2026-07-01', 'business_date_end': '2026-07-31', 'customer_id': 7, 'sub_customer_id': 9, 'verification_status': -1, 'document_type': 8, 'cost_type_id': 12, }, client.calls[-1][2], ) def test_cli_forwards_export_filters(self): client = RecordingApiClient() app = GatewayApp(api_client=client) code = app.run_cli([ 'call', '--tool', 'export_receivable_cost_list', '--order-numbers', 'ORDER-1', ], stdout=io.StringIO()) self.assertEqual(0, code) self.assertEqual({'order_numbers': ['ORDER-1']}, client.calls[-1][2]) code = app.run_cli([ 'call', '--tool', 'export_receivable_cost_list', '--order-numbers', 'ORDER-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', '8', '--cost-type-id', '12', ], stdout=io.StringIO()) self.assertEqual(0, code) self.assertEqual('export_receivable_cost_list', client.calls[-1][0]) self.assertEqual( { 'order_numbers': ['ORDER-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': 8, 'cost_type_id': 12, }, client.calls[-1][2], ) def test_local_and_public_registries_include_export_tool(self): local = GatewayApp().registered_tool_names() public = PublicGatewayApp(None, None).registered_tool_names() self.assertEqual(local, public) self.assertEqual(22, len(local)) self.assertIn('export_receivable_cost_list', local) self.assertEqual(21, len(OutputPresenter.SAFE_TOOLS)) def test_presenter_reuses_queued_export_contract(self): presented = OutputPresenter().present( 'export_receivable_cost_list', { 'code': 'MCP_0000', 'data': { 'task_ref': 'mexp_test', 'status': 'queued', 'retry_after_seconds': 10, }, }, ) self.assertFalse(presented['is_error']) task = presented['structured_content']['task'] self.assertEqual('queued', task['status']) self.assertEqual('mexp_test', task['task_ref']) self.assertEqual(10, task['retry_after_seconds']) def test_wrong_route_path_fails_and_restore_passes(self): tool = ExportReceivableCostListTool() original = tool.route_path tool.route_path = '/mcp/tools/exportReceivableCostListWrong' self.assertNotEqual( '/mcp/tools/exportReceivableCostList', tool.route_path, ) tool.route_path = original self.assertEqual( '/mcp/tools/exportReceivableCostList', tool.route_path, ) if __name__ == '__main__': unittest.main()