mcp/tests/test_app_coverage.py

"""
Coverage补充测试：app.py

目标路径：
- GatewayApp.from_config — token_store_type 不支持 → ValueError
- GatewayApp.run_cli call — else 分支（tool 既非 query_order 也非 query_track，有 keyword）
- GatewayApp.run_cli call — else 分支（tool 既非 query_order 也非 query_track，无 keyword）
- main() — 正常调用 list-tools
"""
import io
import json
import os
import tempfile
import unittest
from unittest.mock import MagicMock, patch

from app import GatewayApp, main
from services.token_store import FileTokenStore
from tools.query_order import QueryOrderTool


# ---------------------------------------------------------------------------
# GatewayApp.from_config — 不支持的 store type
# ---------------------------------------------------------------------------

class FromConfigUnsupportedStoreTest(unittest.TestCase):
    def test_raises_value_error_for_unknown_store_type(self):
        config = MagicMock()
        config.token_store_type = 'memcached'
        with self.assertRaises(ValueError) as ctx:
            GatewayApp.from_config(config)
        self.assertIn('unsupported token store type', str(ctx.exception))
        self.assertIn('memcached', str(ctx.exception))

    def test_from_config_file_store(self):
        """file store 分支：正常返回 GatewayApp 实例。"""
        with tempfile.TemporaryDirectory() as tmp_dir:
            config = MagicMock()
            config.token_store_type = 'file'
            config.token_store_path = os.path.join(tmp_dir, 'tok.json')
            config.refresh_skew_seconds = 120
            config.auth_base_url = 'http://auth.test'
            config.client_type = 'workbuddy'
            config.timeout_seconds = 5
            config.session_key = 'test_session'
            config.tools_base_url = 'http://api.test'
            app = GatewayApp.from_config(config)
            self.assertIsInstance(app, GatewayApp)

    def test_from_config_redis_store_uses_provided_redis_client(self):
        """redis store 分支：传入 redis_client 时跳过 RedisSocketClient 创建。"""
        mock_redis = MagicMock()
        config = MagicMock()
        config.token_store_type = 'redis'
        config.redis_prefix = 'test:'
        config.session_key = 'sess_1'
        config.refresh_skew_seconds = 120
        config.auth_base_url = 'http://auth.test'
        config.client_type = 'workbuddy'
        config.timeout_seconds = 5
        config.tools_base_url = 'http://api.test'
        # get() 返回 None，避免 RedisTokenStore 在初始化时崩溃
        mock_redis.get.return_value = None
        app = GatewayApp.from_config(config, redis_client=mock_redis)
        self.assertIsInstance(app, GatewayApp)


# ---------------------------------------------------------------------------
# GatewayApp.run_cli call — else 分支
# ---------------------------------------------------------------------------

class DummyApiClient:
    def list_enabled_tools(self):
        return {
            'code': 'MCP_0000',
            'data': {
                'tool_codes': [
                    'query_order',
                    'query_track',
                    'query_order_exact',
                    'list_order_filter_options',
                    'fake_tool',
                    'no_kw_tool',
                ],
            },
        }

    def call_tool(self, tool_code, route_path, payload, request_id):
        return {
            'code': 'MCP_0000',
            'data': {},
            'meta': {'request_id': request_id},
        }


class DummyAuthClient:
    def __init__(self, token_store):
        self.token_store = token_store

    def refresh(self, mcp_token):
        pass


def _make_app_with_token():
    """创建有有效 token 的 GatewayApp，附带一个自定义 fake_tool 工具。"""
    with tempfile.TemporaryDirectory() as tmp_dir:
        path = os.path.join(tmp_dir, 'token.json')
        token_store = FileTokenStore(path)
        token_store.save('MT_test', '2099-01-01T00:00:00')
        api_client = DummyApiClient()
        app = GatewayApp(
            auth_client=DummyAuthClient(token_store),
            api_client=api_client,
            token_store=token_store,
        )
        # 注册一个假工具，使 else 分支可以正常 call_tool
        app._tools['fake_tool'] = QueryOrderTool(api_client=api_client)
        return app, tmp_dir


class RunCliCallElseBranchTest(unittest.TestCase):
    def test_call_else_branch_with_keyword_passes_keyword_to_tool(self):
        """
        --tool fake_tool（非 query_order / query_track）且 --keyword 有值 →
        走 else 分支，tool_args 中加入 keyword。
        """
        with tempfile.TemporaryDirectory() as tmp_dir:
            path = os.path.join(tmp_dir, 'token.json')
            token_store = FileTokenStore(path)
            token_store.save('MT_test', '2099-01-01T00:00:00')
            api_client = DummyApiClient()
            app = GatewayApp(
                auth_client=DummyAuthClient(token_store),
                api_client=api_client,
                token_store=token_store,
            )
            app._tools['fake_tool'] = QueryOrderTool(api_client=api_client)

            stdout = io.StringIO()
            code = app.run_cli(
                ['call', '--tool', 'fake_tool', '--keyword', 'hello'],
                stdout=stdout,
            )
            self.assertEqual(0, code)
            payload = json.loads(stdout.getvalue())
            self.assertEqual('MCP_0000', payload['code'])

    def test_call_else_branch_without_keyword_still_calls_tool(self):
        """
        --tool fake_tool（非 query_order / query_track）且 --keyword 为空 →
        走 else 分支，tool_args 中不含 keyword（由工具自己决定是否报错）。
        注：fake_tool 是 QueryOrderTool，无 keyword 会抛 ValueError →
        GatewayApp.run_cli 不 catch，直接抛出。
        """
        with tempfile.TemporaryDirectory() as tmp_dir:
            path = os.path.join(tmp_dir, 'token.json')
            token_store = FileTokenStore(path)
            token_store.save('MT_test', '2099-01-01T00:00:00')
            api_client = DummyApiClient()
            app = GatewayApp(
                auth_client=DummyAuthClient(token_store),
                api_client=api_client,
                token_store=token_store,
            )

            # 注册一个不需要 keyword 的假工具
            class NoKeywordTool:
                name = 'no_kw_tool'
                route_path = '/fake'
                requires_session = False

                def metadata(self):
                    return {'name': self.name, 'description': '', 'input_schema': {'type': 'object', 'properties': {}}}

                def call(self, request_id='', **_kwargs):
                    return {'code': 'MCP_0000', 'data': {}}

            app._tools['no_kw_tool'] = NoKeywordTool()

            stdout = io.StringIO()
            # --keyword 不传，else 分支 keyword 为空 → 不加入 tool_args
            code = app.run_cli(
                ['call', '--tool', 'no_kw_tool'],
                stdout=stdout,
            )
            self.assertEqual(0, code)


# ---------------------------------------------------------------------------
# main() — 正常调用
# ---------------------------------------------------------------------------

class MainFunctionTest(unittest.TestCase):
    def test_main_list_tools_returns_zero(self):
        """
        main() 读取 env、构建 app、调用 run_cli(['list-tools'])。
        直接 mock GatewayConfig.from_env 返回 file store 配置，避免依赖 .env 文件。
        """
        import config as config_module
        with tempfile.TemporaryDirectory() as tmp_dir:
            token_path = os.path.join(tmp_dir, 'token.json')
            fake_config = config_module.GatewayConfig(
                auth_base_url='http://auth.example.com',
                tools_base_url='http://api.example.com',
                token_store_type='file',
                token_store_path=token_path,
                session_key='test_machine:test_user',
            )
            stdout = io.StringIO()
            enabled = {
                'code': 'MCP_0000',
                'data': {
                    'tool_codes': [
                        'query_order',
                        'query_track',
                        'query_order_exact',
                        'list_order_filter_options',
                    ],
                },
            }
            with patch.object(config_module.GatewayConfig, 'from_env', return_value=fake_config):
                with patch('services.api_client.ApiClient.list_enabled_tools', return_value=enabled):
                    with patch('sys.stdout', stdout):
                        code = main(['list-tools'])

            self.assertEqual(0, code)
            tools = json.loads(stdout.getvalue())
            names = [t['name'] for t in tools]
            self.assertIn('query_order', names)
            self.assertIn('query_track', names)
            self.assertIn('query_order_exact', names)
            self.assertIn('list_order_filter_options', names)

    def test_main_propagates_value_error_for_bad_store(self):
        """
        当 token_store_type 为不支持的值时，main() 应抛出 ValueError。
        """
        import config as config_module
        bad_config = config_module.GatewayConfig(
            auth_base_url='http://auth.example.com',
            tools_base_url='http://api.example.com',
            token_store_type='bad_store_type',
            session_key='host:user',
        )
        with self.assertRaises(ValueError):
            with patch.object(config_module.GatewayConfig, 'from_env', return_value=bad_config):
                main(['list-tools'])


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