import importlib import io import math import unittest from app import GatewayApp from public_gateway import PublicGatewayApp from services.output_presenter import OutputPresenter COLUMNS = [ ('order_number', '订单号'), ('customer_name', '客户名称'), ('packing_type', '货物类型'), ('inbound_volume', '入库体积'), ('inbound_date', '入库时间'), ('sales_name', '商务经理'), ('merchandiser_name', '客户经理'), ('department_name', '事业部'), ('product_name', '物流产品'), ] FILTER_TYPES = ['客户名称', '商务经理', '客户经理', '货物类型', '物流产品'] class RecordingApiClient: def __init__(self): self.calls = [] def list_enabled_tools(self, request_id=''): return { 'code': 'MCP_0000', 'data': { 'tool_codes': [ 'query_receive_volume_list', 'list_receive_volume_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': {}, 'meta': {}} class ReceiveVolumeToolContractTest(unittest.TestCase): def tool_class(self, module_name, class_name): return getattr(importlib.import_module('tools.' + module_name), class_name) def test_query_schema_requires_inbound_dates_and_forbids_identity(self): cls = self.tool_class( 'query_receive_volume_list', 'QueryReceiveVolumeListTool' ) metadata = cls().metadata() schema = metadata['input_schema'] self.assertEqual('query_receive_volume_list', metadata['name']) self.assertEqual('/mcp/tools/queryReceiveVolumeList', cls.route_path) self.assertFalse(schema['additionalProperties']) self.assertEqual( ['inbound_date_start', 'inbound_date_end'], schema['required'], ) for forbidden in ( 'company_id', 'admin_id', 'is_super', 'order_numbers', 'business_date_start', ): self.assertNotIn(forbidden, schema['properties']) self.assertIn('使用场景：', metadata['description']) self.assertIn('禁止使用：', metadata['description']) self.assertIn('入库时间', metadata['description']) self.assertIn('admin/Report/receiveForm', metadata['description']) self.assertIn('list_receive_volume_filter_options', metadata['description']) self.assertIn('不得改走query_order', metadata['description']) def test_query_forwards_optional_filters(self): cls = self.tool_class( 'query_receive_volume_list', 'QueryReceiveVolumeListTool' ) client = RecordingApiClient() cls(client).call( inbound_date_start='2026-09-01', inbound_date_end='2026-09-30', customer_ids=[11, 11, 12], sales_ids=[3], merchandiser_ids=[4], packing_types=[' 散货 ', '散货'], product_ids=[8], page=2, limit=30, request_id='rq_rv', ) self.assertEqual( ( 'query_receive_volume_list', '/mcp/tools/queryReceiveVolumeList', { 'inbound_date_start': '2026-09-01', 'inbound_date_end': '2026-09-30', 'customer_ids': [11, 12], 'sales_ids': [3], 'merchandiser_ids': [4], 'packing_types': ['散货'], 'product_ids': [8], 'page': 2, 'limit': 30, }, 'rq_rv', ), client.calls[-1], ) cls(client).call( inbound_date_start='2026-09-01', inbound_date_end='2026-09-01', ) self.assertEqual( { 'inbound_date_start': '2026-09-01', 'inbound_date_end': '2026-09-01', 'page': 1, 'limit': 20, }, client.calls[-1][2], ) with self.assertRaises(RuntimeError): cls().call( inbound_date_start='2026-09-01', inbound_date_end='2026-09-30', ) def test_query_rejects_invalid_window_and_ids(self): cls = self.tool_class( 'query_receive_volume_list', 'QueryReceiveVolumeListTool' ) client = RecordingApiClient() with self.assertRaises(ValueError): cls(client).call( inbound_date_start='2026-09-30', inbound_date_end='2026-09-01', ) with self.assertRaises(ValueError): cls(client).call( inbound_date_start='2026-08-01', inbound_date_end='2026-09-01', ) with self.assertRaises(ValueError): cls(client).call( inbound_date_start='2026-09-01', inbound_date_end='2026-09-30', customer_ids=[0], ) with self.assertRaises(ValueError): cls(client).call( inbound_date_start='2026-09-01', inbound_date_end='2026-09-30', packing_types=[''], ) with self.assertRaises(ValueError): cls(client).call( inbound_date_start='2026-09-01', inbound_date_end='2026-09-30', sales_ids=list(range(1, 202)), ) with self.assertRaises(ValueError): cls(client).call( inbound_date_start='2026-09-01', inbound_date_end='2026-09-30', packing_types=['x' * 101], ) with self.assertRaises(ValueError): cls(client).call( inbound_date_start='2026-09-01', inbound_date_end='2026-09-30', packing_types=[1], ) with self.assertRaises(ValueError): cls(client).call( inbound_date_start='2026-09-01', inbound_date_end='2026-09-30', packing_types=[], ) with self.assertRaises(ValueError): cls(client).call( inbound_date_start='bad', inbound_date_end='2026-09-30', ) with self.assertRaises(ValueError): cls(client).call( inbound_date_start='2026-9-01', inbound_date_end='2026-09-30', ) with self.assertRaises(ValueError): cls(client).call( inbound_date_start='2026-09-01', inbound_date_end='2026-09-30', packing_types=['n{0}'.format(i) for i in range(201)], ) with self.assertRaises(ValueError): cls(client).call( inbound_date_start='2026-09-01', inbound_date_end='2026-09-30', product_ids=['8'], ) with self.assertRaises(ValueError): cls(client).call( inbound_date_start='2026-09-01', inbound_date_end='2026-09-30', merchandiser_ids=[], ) with self.assertRaises(ValueError): cls(client).call( inbound_date_start='2026-09-01', inbound_date_end='2026-09-30', page=101, ) with self.assertRaises(ValueError): cls(client).call( inbound_date_start='2026-09-01', inbound_date_end='2026-09-30', limit=True, ) def test_filter_schema_and_call(self): cls = self.tool_class( 'list_receive_volume_filter_options', 'ListReceiveVolumeFilterOptionsTool', ) metadata = cls().metadata() schema = metadata['input_schema'] self.assertEqual('list_receive_volume_filter_options', metadata['name']) self.assertEqual( '/mcp/tools/listReceiveVolumeFilterOptions', cls.route_path ) self.assertEqual(['filter_type'], schema['required']) self.assertEqual(FILTER_TYPES, schema['properties']['filter_type']['enum']) self.assertIn('query_receive_volume_list', metadata['description']) self.assertIn('禁止猜测', metadata['description']) self.assertIn('使用场景：', metadata['description']) self.assertIn('禁止使用：', metadata['description']) client = RecordingApiClient() cls(client).call('货物类型', keyword='散', page=2, limit=10) self.assertEqual( { 'filter_type': '货物类型', 'keyword': '散', 'page': 2, 'limit': 10, }, client.calls[-1][2], ) with self.assertRaises(ValueError): cls(client).call('事业部') with self.assertRaises(RuntimeError): cls().call('客户名称') with self.assertRaises(ValueError): cls(client).call('客户名称', keyword='x' * 101) with self.assertRaises(ValueError): cls(client).call('客户名称', page=0) with self.assertRaises(ValueError): cls(client).call('客户名称', page=True) def test_cli_forwards_dates_and_filters(self): client = RecordingApiClient() app = GatewayApp(api_client=client) with self.assertRaises(ValueError): app.run_cli([ 'call', '--tool', 'query_receive_volume_list', ], stdout=io.StringIO()) app.run_cli([ 'call', '--tool', 'query_receive_volume_list', '--inbound-date-start', '2026-09-01', '--inbound-date-end', '2026-09-30', '--customer-ids', '11,12', '--sales-ids', '3', '--merchandiser-ids', '4', '--packing-types', '散货,整柜', '--product-ids', '8', ], stdout=io.StringIO()) self.assertEqual( { 'inbound_date_start': '2026-09-01', 'inbound_date_end': '2026-09-30', 'customer_ids': [11, 12], 'sales_ids': [3], 'merchandiser_ids': [4], 'packing_types': ['散货', '整柜'], 'product_ids': [8], 'page': 1, 'limit': 20, }, client.calls[-1][2], ) app.run_cli([ 'call', '--tool', 'query_receive_volume_list', '--inbound-date-start', '2026-09-01', '--inbound-date-end', '2026-09-02', ], stdout=io.StringIO()) self.assertEqual( { 'inbound_date_start': '2026-09-01', 'inbound_date_end': '2026-09-02', 'page': 1, 'limit': 20, }, client.calls[-1][2], ) app.run_cli([ 'call', '--tool', 'list_receive_volume_filter_options', '--filter-type', '物流产品', ], stdout=io.StringIO()) self.assertEqual( { 'filter_type': '物流产品', 'keyword': '', 'page': 1, 'limit': 20, }, client.calls[-1][2], ) def test_presenter_accepts_nine_columns_and_finite_volume(self): presenter = OutputPresenter() columns = [{'key': key, 'name': name} for key, name in COLUMNS] record = { 'order_number': 'FMS-1', 'customer_name': '客户甲', 'packing_type': '散货', 'inbound_volume': 12.5, 'inbound_date': '2026-09-01', 'sales_name': '张三', 'merchandiser_name': '李四', 'department_name': '事业部A', 'product_name': '美线-US01', } meta = { 'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_x', } ok = presenter.present('query_receive_volume_list', { 'code': 'MCP_0000', 'data': {'columns': columns, 'records': [record]}, 'meta': meta, }) self.assertFalse(ok['is_error']) self.assertEqual(12.5, ok['structured_content']['rows'][0][3]) string_volume = dict(record) string_volume['inbound_volume'] = '12.5' self.assertTrue(presenter.present('query_receive_volume_list', { 'code': 'MCP_0000', 'data': {'columns': columns, 'records': [string_volume]}, 'meta': meta, })['is_error']) inf_volume = dict(record) inf_volume['inbound_volume'] = math.inf self.assertTrue(presenter.present('query_receive_volume_list', { 'code': 'MCP_0000', 'data': {'columns': columns, 'records': [inf_volume]}, 'meta': meta, })['is_error']) extra = dict(record) extra['inbound_pieces'] = '1' self.assertTrue(presenter.present('query_receive_volume_list', { 'code': 'MCP_0000', 'data': {'columns': columns, 'records': [extra]}, 'meta': meta, })['is_error']) self.assertTrue(presenter.present('query_receive_volume_list', { 'code': 'MCP_0000', 'data': {'columns': columns, 'records': [], 'extra': 1}, 'meta': meta, })['is_error']) bad_columns = list(columns) bad_columns[0] = {'key': 'order_number', 'name': '错'} self.assertTrue(presenter.present('query_receive_volume_list', { 'code': 'MCP_0000', 'data': {'columns': bad_columns, 'records': []}, 'meta': meta, })['is_error']) non_string = dict(record) non_string['order_number'] = 1 self.assertTrue(presenter.present('query_receive_volume_list', { 'code': 'MCP_0000', 'data': {'columns': columns, 'records': [non_string]}, 'meta': meta, })['is_error']) self.assertTrue(presenter.present('query_receive_volume_list', { 'code': 'MCP_0000', 'data': {'columns': columns, 'records': []}, 'meta': { 'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_x', 'total': 1, }, })['is_error']) options = presenter.present('list_receive_volume_filter_options', { 'code': 'MCP_0000', 'data': { 'records': [ {'value': '散货', 'label': '散货', 'code': ''}, ], }, 'meta': meta, }) self.assertFalse(options['is_error']) def test_local_public_registry_includes_both_tools(self): local = GatewayApp().registered_tool_names() public = PublicGatewayApp(None, None).registered_tool_names() self.assertEqual(local, public) self.assertEqual(32, len(local)) self.assertEqual(31, len(OutputPresenter.SAFE_TOOLS)) self.assertIn('query_receive_volume_list', local) self.assertIn('list_receive_volume_filter_options', local)