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 types import SimpleNamespace
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, request_id=''):
        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


class GatewayAppBoundaryTest(unittest.TestCase):
    def test_protocol_handler_uses_gateway_reporter(self):
        reporter = MagicMock()
        app = GatewayApp(reporter=reporter)

        self.assertIs(reporter, app.create_protocol_handler().reporter)

    def test_registered_tool_names_returns_local_registry(self):
        app = GatewayApp(api_client=DummyApiClient())

        self.assertEqual(tuple(app._tools.keys()), app.registered_tool_names())

    def test_enabled_tool_names_rejects_non_dict_response(self):
        with self.assertRaisesRegex(RuntimeError, 'invalid enabled tool response'):
            GatewayApp()._enabled_tool_names(None)

    def test_enabled_tool_names_rejects_missing_tool_code_list(self):
        response = {'code': 'MCP_0000', 'data': {}}

        with self.assertRaisesRegex(RuntimeError, 'invalid enabled tool response'):
            GatewayApp()._enabled_tool_names(response)

    def test_load_enabled_tools_requires_capable_client(self):
        for api_client in (None, object()):
            with self.subTest(api_client=api_client):
                with self.assertRaisesRegex(RuntimeError, 'client unavailable'):
                    GatewayApp(api_client=api_client)._load_enabled_tool_names()

    def test_build_request_id_preserves_provided_value(self):
        self.assertEqual(
            'rq_given',
            GatewayApp().build_request_id(' rq_given '),
        )

    def test_ensure_session_requires_auth_client_for_expiring_token(self):
        token_store = MagicMock()
        token_store.get.return_value = {'token': 'MT_test'}
        token_store.is_expiring.return_value = True

        with self.assertRaisesRegex(RuntimeError, 'auth client missing'):
            GatewayApp(token_store=token_store).ensure_session()


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'])


class RunCliBranchCoverageTest(unittest.TestCase):
    def _run_with_mocked_call(self, argv):
        app = GatewayApp(api_client=DummyApiClient())
        stdout = io.StringIO()
        with patch.object(
            app,
            'call_tool',
            return_value={'code': 'MCP_0000'},
        ) as call_tool:
            result = app.run_cli(argv, stdout=stdout)
        return result, call_tool

    def test_query_order_requires_keyword(self):
        with self.assertRaisesRegex(ValueError, 'keyword is required'):
            GatewayApp().run_cli([
                'call', '--tool', 'query_order',
            ], stdout=io.StringIO())

    def test_query_track_forwards_each_supported_identifier(self):
        cases = (
            (['--order-id', '5'], {'order_id': 5}),
            (['--order-number', 'ORDER-1'], {'order_number': 'ORDER-1'}),
            (['--tracking-number', 'TRACK-1'], {'tracking_number': 'TRACK-1'}),
        )
        for cli_args, expected in cases:
            with self.subTest(cli_args=cli_args):
                result, call_tool = self._run_with_mocked_call([
                    'call', '--tool', 'query_track', *cli_args,
                ])
                self.assertEqual(0, result)
                payload = call_tool.call_args.args[1]
                for key, value in expected.items():
                    self.assertEqual(value, payload[key])

    def test_query_track_requires_one_identifier(self):
        with self.assertRaisesRegex(ValueError, 'order-id'):
            GatewayApp().run_cli([
                'call', '--tool', 'query_track',
            ], stdout=io.StringIO())

    def test_query_order_exact_forwards_scalar_ids(self):
        result, call_tool = self._run_with_mocked_call([
            'call', '--tool', 'query_order_exact',
            '--order-number', 'ORDER-1',
            '--sales-id', '7',
            '--department-id', '8',
        ])

        self.assertEqual(0, result)
        payload = call_tool.call_args.args[1]
        self.assertEqual(7, payload['sales_id'])
        self.assertEqual(8, payload['department_id'])

    def test_query_order_detail_requires_order_number_and_forwards_section(self):
        with self.assertRaisesRegex(ValueError, 'order-number is required'):
            GatewayApp().run_cli([
                'call', '--tool', 'query_order_detail',
            ], stdout=io.StringIO())

        result, call_tool = self._run_with_mocked_call([
            'call', '--tool', 'query_order_detail',
            '--order-number', 'ORDER-1', '--section', '附件信息',
            '--page', '2', '--limit', '10',
        ])
        self.assertEqual(0, result)
        self.assertEqual({
            'page': 2, 'limit': 10,
            'order_number': 'ORDER-1', 'section': '附件信息',
        }, call_tool.call_args.args[1])

        result, call_tool = self._run_with_mocked_call([
            'call', '--tool', 'query_order_detail', '--order-number', 'ORDER-1',
        ])
        self.assertEqual(0, result)
        self.assertEqual('全部', call_tool.call_args.args[1]['section'])

    def test_filter_options_requires_filter_type(self):
        with self.assertRaisesRegex(ValueError, 'filter-type is required'):
            GatewayApp().run_cli([
                'call', '--tool', 'list_order_filter_options',
            ], stdout=io.StringIO())

    def test_serve_public_builds_scoped_dependencies(self):
        config = SimpleNamespace(
            redis_host='redis.test',
            redis_port=6380,
            redis_db=2,
            redis_password='secret',
            timeout_seconds=9,
            redis_prefix='gateway:',
            gateway_session_ttl_seconds=600,
            tools_base_url='https://tools.test',
            rate_limit_enabled=True,
            rate_limit_max_requests=12,
            rate_limit_window_seconds=34,
            max_in_flight_per_tool=2,
        )
        with patch('app.GatewayConfig.from_env', return_value=config), \
                patch('app.RedisSocketClient') as redis_cls, \
                patch('app.GatewaySessionStore') as store_cls, \
                patch('app.ScopedApiClient') as api_cls, \
                patch('app.PublicGatewayApp') as public_app_cls, \
                patch('app.diagnostic_reporter_from_config') as reporter_factory, \
                patch('app.serve_public', return_value=17) as serve:
            result = GatewayApp().run_cli([
                'serve-public', '--host', '127.0.0.1', '--port', '9000',
            ])

        self.assertEqual(17, result)
        redis_cls.assert_called_once_with(
            host='redis.test',
            port=6380,
            db=2,
            password='secret',
            timeout=9,
        )
        store_cls.assert_called_once_with(
            redis_cls.return_value,
            prefix='gateway:',
            ttl_seconds=600,
        )
        api_cls.assert_called_once_with('https://tools.test', timeout=9)
        public_app_cls.assert_called_once_with(
            session_store=store_cls.return_value,
            api_client=api_cls.return_value,
        )
        serve.assert_called_once_with(
            public_app_cls.return_value,
            host='127.0.0.1',
            port=9000,
            enable_rate_limit=True,
            rate_limit_max_requests=12,
            rate_limit_window_seconds=34,
            max_in_flight_per_tool=2,
            reporter=reporter_factory.return_value,
        )
        reporter_factory.return_value.close.assert_called_once_with()

    def test_serve_public_reuses_existing_non_null_reporter(self):
        config = SimpleNamespace(
            redis_host='redis.test',
            redis_port=6379,
            redis_db=0,
            redis_password='',
            timeout_seconds=1,
            redis_prefix='gateway:',
            gateway_session_ttl_seconds=600,
            tools_base_url='https://tools.test',
            rate_limit_enabled=False,
            rate_limit_max_requests=0,
            rate_limit_window_seconds=60,
            max_in_flight_per_tool=1,
        )
        reporter = MagicMock()
        with patch('app.GatewayConfig.from_env', return_value=config), \
                patch('app.RedisSocketClient'), \
                patch('app.GatewaySessionStore'), \
                patch('app.ScopedApiClient'), \
                patch('app.PublicGatewayApp'), \
                patch('app.diagnostic_reporter_from_config') as factory, \
                patch('app.serve_public', return_value=0) as serve:
            result = GatewayApp(reporter=reporter).run_cli([
                'serve-public', '--host', '127.0.0.1', '--port', '9000',
            ])

        self.assertEqual(0, result)
        factory.assert_not_called()
        self.assertIs(reporter, serve.call_args.kwargs['reporter'])
        reporter.close.assert_called_once_with()

    def test_unsupported_command_raises_runtime_error(self):
        with patch(
            'app.argparse.ArgumentParser.parse_args',
            return_value=SimpleNamespace(command='unknown'),
        ):
            with self.assertRaisesRegex(RuntimeError, 'unsupported command'):
                GatewayApp().run_cli([], stdout=io.StringIO())

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