import importlib import io import unittest from app import GatewayApp from public_gateway import PublicGatewayApp from services.output_presenter import OutputPresenter PROGRESS_COLUMNS = [ ('bl_number', '提单号'), ('container_code', '柜号'), ('container_type', '柜型'), ('warehouse_name', '仓库名称'), ('providers_name', '拖车行'), ('trailer_status', '提柜状态'), ('latest_remark', '最新操作备注'), ('is_direct_send', '直送'), ('cabinet_type', '整拼类型'), ('vessel_name', '船名航次'), ('route', '航线'), ('clearance_check', '清关查验'), ('etd', 'ETD'), ('atd', 'ATD'), ('eta', 'ETA'), ('ata', 'ATA'), ('loading_port', '起运港'), ('destination_port', '目的港'), ('wharf_name', '到港码头'), ('appt_time', 'APPT时间'), ('delivery_end_time', '卡车实际派送时间'), ('est_pickup_time', '预计提柜时间'), ('pickup_time', '实际提柜时间'), ('inbound_date', '到仓时间'), ('container_return_time', '还柜时间'), ('pickup_prescription', '提柜时效'), ('return_prescription', '还柜时效'), ('wharf_wait_time', '码头等待时间'), ('amazon_wait_time', '亚马逊等待时间'), ('bill_status', '应付费用状态'), ] DO_COLUMNS = [ ('trailer_number', '拖车单号'), ('outbound_number', '出库单号'), ('container_code', '柜号'), ('container_type', '柜型'), ('bl_number', '提单号'), ('is_direct_send', '是否直送'), ('original_do_file', '原DO单'), ('new_do_file', '新DO单'), ('email_subject', '邮件主题'), ('email_from', '邮件发送人'), ('received_at', '邮件接收时间'), ] FILTER_TYPES = [ '海外提柜类型', '拖车行', '海外仓', '直送柜', '应付费用', '整拼类型', '运输方式', ] STAGES = [ '待安排拖车', '待提柜', '待还柜(海)', '已还柜(海)', '已提货(空)', 'DO单制作', ] class RecordingApiClient: def __init__(self): self.calls = [] def list_enabled_tools(self, request_id=''): return { 'code': 'MCP_0000', 'data': { 'tool_codes': [ 'query_destination_trailer_list', 'list_destination_trailer_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 DestinationTrailerToolContractTest(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_stage_and_forbids_identity(self): cls = self.tool_class( 'query_destination_trailer_list', 'QueryDestinationTrailerListTool' ) metadata = cls().metadata() schema = metadata['input_schema'] self.assertEqual('query_destination_trailer_list', metadata['name']) self.assertEqual('/mcp/tools/queryDestinationTrailerList', cls.route_path) self.assertFalse(schema['additionalProperties']) self.assertEqual(['trailer_stage'], schema['required']) self.assertEqual(STAGES, schema['properties']['trailer_stage']['enum']) self.assertNotIn('default', schema['properties']['trailer_stage']) for forbidden in ('company_id', 'admin_id', 'is_super', 'outbound_id'): self.assertNotIn(forbidden, schema['properties']) self.assertIn('使用场景：', metadata['description']) self.assertIn('禁止使用：', metadata['description']) self.assertIn('号码类型不明确时必须先询问用户', metadata['description']) self.assertIn('不得根据号码格式猜测', metadata['description']) self.assertIn('不得跨字段或跨工具试查', metadata['description']) self.assertIn('query_outbound_list', metadata['description']) self.assertIn('提单号、柜号、日期和其他筛选均为可选', metadata['description']) self.assertNotIn('必须再有', metadata['description']) def test_query_forwards_numbers_dates_and_rejects_invalid(self): cls = self.tool_class( 'query_destination_trailer_list', 'QueryDestinationTrailerListTool' ) client = RecordingApiClient() cls(client).call( trailer_stage='待提柜', bl_numbers=[' BL-1 ', 'BL-1'], container_codes=['CONT-1'], providers_id=8, warehouse_id=3, is_direct_send=0, bill_status=10, cabinet_type=1, shipping_method=2, eta_start='2026-07-01', eta_end='2026-07-31', page=2, limit=30, request_id='rq_dt', ) self.assertEqual( { 'trailer_stage': '待提柜', 'bl_numbers': ['BL-1'], 'container_codes': ['CONT-1'], 'providers_id': 8, 'warehouse_id': 3, 'is_direct_send': 0, 'bill_status': 10, 'cabinet_type': 1, 'shipping_method': 2, 'eta_start': '2026-07-01', 'eta_end': '2026-07-31', 'page': 2, 'limit': 30, }, client.calls[-1][2], ) self.assertEqual( '/mcp/tools/queryDestinationTrailerList', client.calls[-1][1], ) cls(client).call(trailer_stage='待提柜') self.assertEqual( { 'trailer_stage': '待提柜', 'page': 1, 'limit': 20, }, client.calls[-1][2], ) cls(client).call(trailer_stage='待提柜', providers_id=8) with self.assertRaises(RuntimeError): cls().call(trailer_stage='待提柜') invalid = ( {'trailer_stage': '全部', 'bl_numbers': ['BL-1']}, {'trailer_stage': '待提柜', 'eta_start': '2026-07-01'}, { 'trailer_stage': '待提柜', 'eta_start': '2026-07-01', 'eta_end': '2026-08-01', }, { 'trailer_stage': 'DO单制作', 'bl_numbers': ['BL-1'], 'providers_id': 8, }, { 'trailer_stage': 'DO单制作', 'eta_start': '2026-07-01', 'eta_end': '2026-07-02', }, ) for arguments in invalid: with self.subTest(arguments=arguments): with self.assertRaises(ValueError): cls(client).call(**arguments) def test_filter_schema_and_forwarding(self): cls = self.tool_class( 'list_destination_trailer_filter_options', 'ListDestinationTrailerFilterOptionsTool', ) metadata = cls().metadata() schema = metadata['input_schema'] self.assertEqual(['filter_type'], schema['required']) self.assertEqual(FILTER_TYPES, schema['properties']['filter_type']['enum']) 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('海外提柜类型', limit=True) def test_query_covers_remaining_validation_branches(self): cls = self.tool_class( 'query_destination_trailer_list', 'QueryDestinationTrailerListTool' ) client = RecordingApiClient() with self.assertRaises(ValueError): cls(client).call( trailer_stage='待提柜', bl_numbers=['X'] * 201, ) with self.assertRaises(ValueError): cls(client).call( trailer_stage='待提柜', eta_start='2026-07-02', eta_end='2026-07-01', ) with self.assertRaises(ValueError): cls(client).call( trailer_stage='待提柜', bl_numbers=['BL-1'], bill_status=99, ) with self.assertRaises(ValueError): cls(client).call(trailer_stage='待提柜', bl_numbers='BL-1') with self.assertRaises(ValueError): cls(client).call(trailer_stage='待提柜', bl_numbers=[1]) with self.assertRaises(ValueError): cls(client).call(trailer_stage='待提柜', bl_numbers=[' ']) with self.assertRaises(ValueError): cls(client).call( trailer_stage='待提柜', eta_start=20260701, eta_end='2026-07-02', ) with self.assertRaises(ValueError): cls(client).call( trailer_stage='待提柜', eta_start='2026-99-01', eta_end='2026-07-02', ) with self.assertRaises(ValueError): cls(client).call( trailer_stage='待提柜', eta_start='2026-7-01', eta_end='2026-07-02', ) with self.assertRaises(ValueError): cls(client).call( trailer_stage='待提柜', bl_numbers=['BL-1'], providers_id=True, ) with self.assertRaises(ValueError): cls(client).call( trailer_stage='待提柜', bl_numbers=['BL-1'], warehouse_id=0, ) with self.assertRaises(ValueError): cls(client).call( trailer_stage='待提柜', bl_numbers=['BL-1'], page=0, ) with self.assertRaises(ValueError): cls(client).call( trailer_stage='待提柜', bl_numbers=['BL-1'], limit=True, ) cls(client).call( trailer_stage='待安排拖车', pickup_time_start='2026-07-01', pickup_time_end='2026-07-02', providers_id=8, bill_status=20, ) cls(client).call(trailer_stage='DO单制作', bl_numbers=['BL-1']) with self.assertRaises(ValueError): cls(client).call( trailer_stage='待提柜', eta_start='20260701', eta_end='2026-07-02', ) def test_cli_forwards_optional_trailer_filters(self): client = RecordingApiClient() app = GatewayApp(api_client=client) with self.assertRaises(ValueError): app.run_cli([ 'call', '--tool', 'query_destination_trailer_list', ], stdout=io.StringIO()) app.run_cli([ 'call', '--tool', 'query_destination_trailer_list', '--trailer-stage', '待还柜(海)', '--container-codes', 'C1', '--providers-id', '8', '--warehouse-id', '3', '--is-direct-send', '1', '--bill-status', '30', '--cabinet-type', '2', '--shipping-method', '2', '--eta-start', '2026-07-01', '--eta-end', '2026-07-02', '--pickup-time-start', '2026-07-03', '--pickup-time-end', '2026-07-04', '--container-return-time-start', '2026-07-05', '--container-return-time-end', '2026-07-06', ], stdout=io.StringIO()) self.assertEqual( { 'trailer_stage': '待还柜(海)', 'container_codes': ['C1'], 'providers_id': 8, 'warehouse_id': 3, 'is_direct_send': 1, 'bill_status': 30, 'cabinet_type': 2, 'shipping_method': 2, 'eta_start': '2026-07-01', 'eta_end': '2026-07-02', 'pickup_time_start': '2026-07-03', 'pickup_time_end': '2026-07-04', 'container_return_time_start': '2026-07-05', 'container_return_time_end': '2026-07-06', 'page': 1, 'limit': 20, }, client.calls[-1][2], ) def test_presenter_fails_closed_on_unknown_shape(self): presenter = OutputPresenter() meta = { 'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_x', } self.assertTrue(presenter.present('query_destination_trailer_list', { 'code': 'MCP_0000', 'data': {'trailer_stage': '待提柜', 'columns': [], 'records': [], 'extra': 1}, 'meta': meta, })['is_error']) self.assertTrue(presenter.present('query_destination_trailer_list', { 'code': 'MCP_0000', 'data': { 'trailer_stage': '未知', 'columns': [], 'records': [], }, 'meta': meta, })['is_error']) self.assertTrue(presenter.present('query_destination_trailer_list', { 'code': 'MCP_0000', 'data': { 'trailer_stage': '待提柜', 'columns': [{'key': 'bl_number', 'name': '提单号'}], 'records': [], }, 'meta': meta, })['is_error']) columns = [{'key': key, 'name': name} for key, name in PROGRESS_COLUMNS] columns[0] = {'key': 'bl_number', 'name': '错'} self.assertTrue(presenter.present('query_destination_trailer_list', { 'code': 'MCP_0000', 'data': { 'trailer_stage': '待提柜', 'columns': columns, 'records': [], }, 'meta': meta, })['is_error']) record = {key: 'v' for key, _ in PROGRESS_COLUMNS} record['bl_number'] = 1 self.assertTrue(presenter.present('query_destination_trailer_list', { 'code': 'MCP_0000', 'data': { 'trailer_stage': '待提柜', 'columns': [ {'key': key, 'name': name} for key, name in PROGRESS_COLUMNS ], 'records': [record], }, 'meta': meta, })['is_error']) def test_local_public_registry_cli_and_safe_counts(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_destination_trailer_list', local) self.assertIn('list_destination_trailer_filter_options', local) client = RecordingApiClient() app = GatewayApp(api_client=client) app.run_cli([ 'call', '--tool', 'query_destination_trailer_list', '--trailer-stage', '待提柜', '--bl-numbers', ' BL-1,BL-2 ', '--page', '2', '--limit', '30', ], stdout=io.StringIO()) self.assertEqual( { 'trailer_stage': '待提柜', 'bl_numbers': ['BL-1', 'BL-2'], 'page': 2, 'limit': 30, }, client.calls[-1][2], ) with self.assertRaises(ValueError): app.run_cli([ 'call', '--tool', 'list_destination_trailer_filter_options', ], stdout=io.StringIO()) app.run_cli([ 'call', '--tool', 'list_destination_trailer_filter_options', '--filter-type', '海外提柜类型', ], stdout=io.StringIO()) self.assertEqual( 'list_destination_trailer_filter_options', client.calls[-1][0], ) def test_presenter_locks_thirty_and_eleven_columns(self): presenter = OutputPresenter() progress_record = {key: 'v' for key, _ in PROGRESS_COLUMNS} progress_record['trailer_status'] = '待提柜' result = presenter.present('query_destination_trailer_list', { 'code': 'MCP_0000', 'data': { 'trailer_stage': '待提柜', 'columns': [ {'key': key, 'name': name} for key, name in PROGRESS_COLUMNS ], 'records': [progress_record], }, 'meta': { 'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_p', }, }) self.assertFalse(result['is_error']) self.assertEqual(30, len(result['structured_content']['headers'])) self.assertEqual('待提柜', result['structured_content']['trailer_stage']) do_record = {key: 'd' for key, _ in DO_COLUMNS} do_record['is_direct_send'] = '否' do_record['original_do_file'] = '' do_record['new_do_file'] = 'https://files.example/do.pdf' do_ok = presenter.present('query_destination_trailer_list', { 'code': 'MCP_0000', 'data': { 'trailer_stage': 'DO单制作', 'columns': [ {'key': key, 'name': name} for key, name in DO_COLUMNS ], 'records': [do_record], }, 'meta': { 'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_d', }, }) self.assertFalse(do_ok['is_error']) self.assertEqual(11, len(do_ok['structured_content']['headers'])) bad_url = dict(do_record) bad_url['new_do_file'] = 'javascript:alert(1)' bad = presenter.present('query_destination_trailer_list', { 'code': 'MCP_0000', 'data': { 'trailer_stage': 'DO单制作', 'columns': [ {'key': key, 'name': name} for key, name in DO_COLUMNS ], 'records': [bad_url], }, 'meta': { 'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_b', }, }) self.assertTrue(bad['is_error']) extra = dict(progress_record) extra['outbound_number'] = 'OB1' extra_result = presenter.present('query_destination_trailer_list', { 'code': 'MCP_0000', 'data': { 'trailer_stage': '待提柜', 'columns': [ {'key': key, 'name': name} for key, name in PROGRESS_COLUMNS ], 'records': [extra], }, 'meta': { 'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_e', }, }) self.assertTrue(extra_result['is_error'])