import io import unittest from unittest.mock import patch from app import GatewayApp from public_gateway import PublicGatewayApp from services.output_presenter import OutputPresenter from tools.query_freightower_track import QueryFreightowerTrackTool class RecordingApiClient: def __init__(self): self.call = None def call_tool(self, tool_code, route_path, payload, request_id): self.call = (tool_code, route_path, payload, request_id) return {'code': 'MCP_0000', 'data': {'ok': True}} def sample_data(): return { 'schedule': { 'bl_number': 'BL1', 'container_code': 'BOX1', 'vessel': 'VESSEL', 'voyage': '001E', 'etd': '2026-09-01 00:00:00', 'eta': '2026-09-20 00:00:00', }, 'ports': [{ 'place_type': '目的港', 'port_name': '洛杉矶', 'eta': '2026-09-20', 'ata': '', }], 'tracks': [{ 'time': '2026-09-20 01:00:00', 'location': 'LOS ANGELES', 'content': '抵港', 'event_code': 'BDAR', }], } class QueryFreightowerTrackToolTest(unittest.TestCase): def test_metadata_requires_explicit_number_type(self): metadata = QueryFreightowerTrackTool().metadata() description = metadata['description'] schema = metadata['input_schema'] self.assertEqual('query_freightower_track', metadata['name']) self.assertNotIn('required', schema) self.assertFalse(schema['additionalProperties']) self.assertIn('使用场景：', description) self.assertIn('禁止使用：', description) self.assertIn('号码类型不明确时必须先询问用户', description) self.assertIn('不得根据号码格式猜测', description) self.assertIn('不得跨字段或跨工具试查', description) self.assertIn('query_track', description) self.assertIn('仅当用户明确', schema['properties']['bl_number']['description']) self.assertIn('不得放入其他类型号码', schema['properties']['container_code']['description']) def test_call_forwards_only_provided_numbers(self): api = RecordingApiClient() tool = QueryFreightowerTrackTool(api_client=api) tool.call(bl_number=' BL1 ', container_code=' BOX1 ', request_id='rq_1') self.assertEqual(( 'query_freightower_track', '/mcp/tools/queryFreightowerTrack', {'bl_number': 'BL1', 'container_code': 'BOX1'}, 'rq_1', ), api.call) def test_call_accepts_one_number(self): api = RecordingApiClient() QueryFreightowerTrackTool(api_client=api).call(container_code='BOX1') self.assertEqual({'container_code': 'BOX1'}, api.call[2]) def test_invalid_numbers_are_rejected(self): tool = QueryFreightowerTrackTool(api_client=RecordingApiClient()) with self.assertRaises(ValueError): tool.call() with self.assertRaises(ValueError): tool.call(bl_number=123) with self.assertRaises(ValueError): tool.call(container_code='B' * 101) def test_api_client_is_required(self): with self.assertRaises(RuntimeError): QueryFreightowerTrackTool().call(bl_number='BL1') def test_registries_include_the_tool(self): local = GatewayApp().registered_tool_names() public = PublicGatewayApp(None, None).registered_tool_names() self.assertEqual(local, public) self.assertIn('query_freightower_track', local) self.assertEqual(35, len(local)) self.assertEqual(34, len(OutputPresenter.SAFE_TOOLS)) def test_cli_forwards_bill_and_container(self): app = GatewayApp() stdout = io.StringIO() with patch.object(app, 'call_tool', return_value={'code': 'MCP_0000'}) as call_tool: code = app.run_cli([ 'call', '--tool', 'query_freightower_track', '--bl-number', 'BL1', '--container-code', 'BOX1', ], stdout=stdout) self.assertEqual(0, code) self.assertEqual('query_freightower_track', call_tool.call_args[0][0]) self.assertEqual( {'bl_number': 'BL1', 'container_code': 'BOX1'}, call_tool.call_args[0][1], ) def test_cli_requires_a_number(self): app = GatewayApp() with self.assertRaises(ValueError): app.run_cli([ 'call', '--tool', 'query_freightower_track', ], stdout=io.StringIO()) def test_presenter_renders_schedule_ports_and_arrival(self): result = OutputPresenter().present('query_freightower_track', { 'code': 'MCP_0000', 'data': sample_data(), 'meta': {'request_id': 'rq_ft'}, }) self.assertFalse(result['is_error']) self.assertEqual('洛杉矶', result['structured_content']['港口'][0]['港口']) self.assertEqual('BDAR', result['structured_content']['轨迹'][0]['状态代码']) self.assertIn('抵港', result['text']) self.assertNotIn('bl_number', result['text']) self.assertEqual({'request_id': 'rq_ft'}, result['meta']) def test_presenter_accepts_empty_ports_and_tracks(self): data = sample_data() data['ports'] = [] data['tracks'] = [] result = OutputPresenter().present('query_freightower_track', { 'code': 'MCP_0000', 'data': data, }) self.assertFalse(result['is_error']) self.assertIn('港口\n- 无', result['text']) self.assertIn('轨迹\n- 无', result['text']) def test_presenter_rejects_unknown_shape(self): cases = [] extra = sample_data() extra['schedule']['secret'] = 'x' cases.append(extra) bad_type = sample_data() bad_type['ports'][0]['place_type'] = '起运港' cases.append(bad_type) bad_value = sample_data() bad_value['tracks'][0]['time'] = 1 cases.append(bad_value) cases.append({'schedule': {}}) for data in cases: with self.subTest(data=data): result = OutputPresenter().present('query_freightower_track', { 'code': 'MCP_0000', 'data': data, }) self.assertTrue(result['is_error']) self.assertNotIn('secret', result['text']) def test_presenter_keeps_allowlisted_business_message(self): result = OutputPresenter().present('query_freightower_track', { 'code': 'MCP_1501', 'msg': '暂时没有飞驼轨迹', 'data': [], }) self.assertTrue(result['is_error']) self.assertIn('暂时没有飞驼轨迹', result['text']) def test_presenter_hides_unknown_upstream_message(self): result = OutputPresenter().present('query_freightower_track', { 'code': 'MCP_1501', 'msg': 'carrier token leaked', 'data': [], }) self.assertTrue(result['is_error']) self.assertNotIn('leaked', result['text']) self.assertIn('目标数据不可用', result['text']) if __name__ == '__main__': unittest.main()