|
|
@@ -0,0 +1,654 @@
|
|
|
+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()
|
jackson