mcp/tests/test_query_freightower_track_tool.py

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()