mcp/tests/test_query_export_task_tool.py

import io
import unittest
from unittest.mock import patch

from app import GatewayApp
from public_gateway import PublicGatewayApp
from tools.query_export_task import QueryExportTaskTool


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': {'task_ref': payload['task_ref'], 'status': 'queued'},
        }


class QueryExportTaskToolTest(unittest.TestCase):
    def test_metadata_exposes_only_bounded_task_reference(self):
        metadata = QueryExportTaskTool().metadata()
        schema = metadata['input_schema']

        self.assertEqual('query_export_task', metadata['name'])
        self.assertEqual(['task_ref'], schema['required'])
        self.assertFalse(schema['additionalProperties'])
        self.assertEqual('string', schema['properties']['task_ref']['type'])
        self.assertEqual(1, schema['properties']['task_ref']['minLength'])
        self.assertEqual(512, schema['properties']['task_ref']['maxLength'])
        self.assertIn('单独', metadata['description'])
        self.assertIn('不得在同一次调用内等待', metadata['description'])

    def test_call_trims_and_forwards_reference(self):
        api = RecordingApiClient()
        tool = QueryExportTaskTool(api_client=api)

        result = tool.call(task_ref=' mexp_abc ', request_id='rq_1')

        self.assertEqual('queued', result['data']['status'])
        self.assertEqual((
            'query_export_task',
            '/mcp/tools/queryExportTask',
            {'task_ref': 'mexp_abc'},
            'rq_1',
        ), api.call)

    def test_invalid_references_are_rejected(self):
        tool = QueryExportTaskTool(api_client=RecordingApiClient())
        for value in (None, '', '   ', 123, 'a' * 513):
            with self.subTest(value=value):
                with self.assertRaises(ValueError):
                    tool.call(task_ref=value)

    def test_api_client_is_required(self):
        with self.assertRaises(RuntimeError):
            QueryExportTaskTool().call(task_ref='mexp_abc')

    def test_local_and_public_registries_contain_same_thirteen_tools(self):
        local = GatewayApp(api_client=RecordingApiClient()).registered_tool_names()
        public = PublicGatewayApp(None, None).registered_tool_names()

        self.assertEqual(local, public)
        self.assertEqual(13, len(local))
        self.assertIn('query_export_task', local)

    def test_cli_forwards_only_task_reference(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_export_task',
                '--task-ref', 'mexp_abc',
                '--request-id', 'rq_cli',
            ], stdout=stdout)

        self.assertEqual(0, code)
        call_tool.assert_called_once_with(
            'query_export_task',
            {'task_ref': 'mexp_abc'},
            request_id='rq_cli',
        )

    def test_cli_requires_task_reference(self):
        with self.assertRaises(ValueError):
            GatewayApp().run_cli([
                'call',
                '--tool', 'query_export_task',
            ], stdout=io.StringIO())


if __name__ == '__main__':
    unittest.main()