mcp/tests/test_config_compat.py

import os
import tempfile
import unittest

from config import GatewayConfig


class GatewayConfigCompatTest(unittest.TestCase):
    def test_gateway_config_accepts_documented_fms_env_names_when_dotenv_missing(self):
        env = {
            'FMS_API_BASE': 'http://gateway.example.test',
            'FMS_CLIENT_TYPE': 'workbuddy',
            'FMS_TIMEOUT_MS': '15000',
            'FMS_LOG_LEVEL': 'debug',
        }

        config = GatewayConfig.from_env(env, dotenv_path=os.path.join(tempfile.gettempdir(), 'missing-fms-mcp.env'))

        self.assertEqual('http://gateway.example.test', config.auth_base_url)
        self.assertEqual('http://gateway.example.test', config.tools_base_url)
        self.assertEqual('workbuddy', config.client_type)
        self.assertEqual(15, config.timeout_seconds)
        self.assertEqual('debug', config.log_level)

    def test_gateway_config_reads_dotenv_file_when_env_missing(self):
        with tempfile.TemporaryDirectory() as tmp_dir:
            dotenv_path = os.path.join(tmp_dir, '.env')
            with open(dotenv_path, 'w', encoding='utf-8') as file:
                file.write('FMS_API_BASE=http://dotenv.example.test\n')
                file.write('FMS_CLIENT_TYPE=workbuddy\n')
                file.write('FMS_TIMEOUT_SECONDS=12\n')
                file.write('FMS_LOG_LEVEL=warning\n')
                file.write('FMS_TOKEN_STORE_PATH=.runtime-token.json\n')

            config = GatewayConfig.from_env(env={}, dotenv_path=dotenv_path)

            self.assertEqual('http://dotenv.example.test', config.auth_base_url)
            self.assertEqual('http://dotenv.example.test', config.tools_base_url)
            self.assertEqual('workbuddy', config.client_type)
            self.assertEqual(12, config.timeout_seconds)
            self.assertEqual('warning', config.log_level)
            self.assertEqual('.runtime-token.json', config.token_store_path)

    def test_gateway_config_prefers_dotenv_over_environment(self):
        with tempfile.TemporaryDirectory() as tmp_dir:
            dotenv_path = os.path.join(tmp_dir, '.env')
            with open(dotenv_path, 'w', encoding='utf-8') as file:
                file.write('FMS_AUTH_BASE=http://dotenv-auth.example.test\n')
                file.write('FMS_TOOLS_BASE=http://dotenv-tools.example.test\n')
                file.write('FMS_CLIENT_TYPE=workbuddy-dotenv\n')
                file.write('FMS_TIMEOUT_SECONDS=12\n')
                file.write('FMS_LOG_LEVEL=warning\n')
                file.write('FMS_TOKEN_STORE_PATH=.dotenv-token.json\n')

            config = GatewayConfig.from_env(
                env={
                    'FMS_AUTH_BASE': 'http://env-auth.example.test',
                    'FMS_TOOLS_BASE': 'http://env-tools.example.test',
                    'FMS_CLIENT_TYPE': 'workbuddy-env',
                    'FMS_TIMEOUT_SECONDS': '30',
                    'FMS_LOG_LEVEL': 'debug',
                    'FMS_TOKEN_STORE_PATH': '.env-token.json',
                },
                dotenv_path=dotenv_path,
            )

            self.assertEqual('http://dotenv-auth.example.test', config.auth_base_url)
            self.assertEqual('http://dotenv-tools.example.test', config.tools_base_url)
            self.assertEqual('workbuddy-dotenv', config.client_type)
            self.assertEqual(12, config.timeout_seconds)
            self.assertEqual('warning', config.log_level)
            self.assertEqual('.dotenv-token.json', config.token_store_path)

    def test_public_gateway_config_reads_session_ttl_and_mode(self):
        config = GatewayConfig.from_env(env={
            'FMS_API_BASE': 'https://base.example.com',
            'FMS_GATEWAY_MODE': 'public',
            'FMS_GATEWAY_SESSION_TTL_SECONDS': '600',
            'FMS_REDIS_PREFIX': 'fms:mcp:gateway:',
        }, dotenv_path='missing.env')

        self.assertEqual('public', config.gateway_mode)
        self.assertEqual(600, config.gateway_session_ttl_seconds)
        self.assertEqual('fms:mcp:gateway:', config.redis_prefix)

    def test_public_gateway_mode_defaults_to_gateway_redis_prefix(self):
        config = GatewayConfig.from_env(env={
            'FMS_API_BASE': 'https://base.example.com',
            'FMS_GATEWAY_MODE': 'public',
        }, dotenv_path='missing.env')

        self.assertEqual('public', config.gateway_mode)
        self.assertEqual('fms:mcp:gateway:', config.redis_prefix)

    def test_timeout_ms_conversion_in_preferred_env(self):
        """Test FMS_TIMEOUT_MS conversion in preferred env"""
        config = GatewayConfig.from_env(env={
            'FMS_API_BASE': 'https://base.example.com',
            'FMS_TIMEOUT_MS': '5000',  # 5000ms = 5 seconds
        }, dotenv_path='missing.env')

        self.assertEqual(5, config.timeout_seconds)

    def test_timeout_ms_minimum_value(self):
        """Test that FMS_TIMEOUT_MS converts to at least 1 second"""
        config = GatewayConfig.from_env(env={
            'FMS_API_BASE': 'https://base.example.com',
            'FMS_TIMEOUT_MS': '500',  # 500ms should become 1 second
        }, dotenv_path='missing.env')

        self.assertEqual(1, config.timeout_seconds)

    def test_timeout_resolution_covers_supported_env_keys(self):
        cases = (
            ({'MCP_TIMEOUT_SECONDS': '11'}, {}, 11),
            ({'FMS_TIMEOUT_MS': '2500'}, {}, 2),
            ({}, {'FMS_TIMEOUT_SECONDS': '12'}, 12),
        )
        for preferred, fallback, expected in cases:
            with self.subTest(preferred=preferred, fallback=fallback):
                self.assertEqual(
                    expected,
                    GatewayConfig._resolve_timeout_seconds(
                        preferred,
                        fallback,
                    ),
                )

    def test_load_dotenv_skips_invalid_lines_and_unquotes_values(self):
        with tempfile.TemporaryDirectory() as tmp_dir:
            path = os.path.join(tmp_dir, '.env')
            with open(path, 'w', encoding='utf-8') as file:
                file.write('# comment\n')
                file.write('invalid-line\n')
                file.write('=ignored\n')
                file.write('DOUBLE="double value"\n')
                file.write("SINGLE='single value'\n")

            values = GatewayConfig._load_dotenv(path)

        self.assertEqual({
            'DOUBLE': 'double value',
            'SINGLE': 'single value',
        }, values)

class RateLimitConfigTest(unittest.TestCase):
    """Tests for _parse_int / _parse_bool helpers and rate limit config parsing."""

    def _config_from_env(self, env):
        return GatewayConfig.from_env(
            env=dict({'FMS_API_BASE': 'http://x.test'}, **env),
            dotenv_path='missing.env',
        )

    # --- _parse_int via OS env var ---

    def test_rate_limit_max_requests_reads_from_env(self):
        config = self._config_from_env({'FMS_RATE_LIMIT_MAX_REQUESTS': '120'})
        self.assertEqual(120, config.rate_limit_max_requests)

    def test_rate_limit_window_seconds_reads_from_env(self):
        config = self._config_from_env({'FMS_RATE_LIMIT_WINDOW_SECONDS': '30'})
        self.assertEqual(30, config.rate_limit_window_seconds)

    def test_max_in_flight_per_tool_reads_from_env(self):
        config = self._config_from_env({'FMS_MAX_IN_FLIGHT_PER_TOOL': '3'})
        self.assertEqual(3, config.max_in_flight_per_tool)

    def test_rate_limit_max_requests_os_env_with_inline_comment_does_not_crash(self):
        # OS env vars are not processed by _load_dotenv, so inline comments must be
        # stripped by _parse_int before int() conversion
        config = self._config_from_env({'FMS_RATE_LIMIT_MAX_REQUESTS': '60 # max per window'})
        self.assertEqual(60, config.rate_limit_max_requests)

    def test_rate_limit_window_seconds_invalid_value_falls_back_to_default(self):
        config = self._config_from_env({'FMS_RATE_LIMIT_WINDOW_SECONDS': 'not_a_number'})
        self.assertEqual(60, config.rate_limit_window_seconds)

    def test_rate_limit_max_requests_empty_falls_back_to_default(self):
        config = self._config_from_env({})
        self.assertEqual(60, config.rate_limit_max_requests)

    # --- _parse_bool via OS env var ---

    def test_rate_limit_enabled_default_is_true(self):
        config = self._config_from_env({})
        self.assertTrue(config.rate_limit_enabled)

    def test_rate_limit_enabled_zero_disables(self):
        config = self._config_from_env({'FMS_RATE_LIMIT_ENABLED': '0'})
        self.assertFalse(config.rate_limit_enabled)

    def test_rate_limit_enabled_false_string_disables(self):
        config = self._config_from_env({'FMS_RATE_LIMIT_ENABLED': 'false'})
        self.assertFalse(config.rate_limit_enabled)

    def test_rate_limit_enabled_off_string_disables(self):
        config = self._config_from_env({'FMS_RATE_LIMIT_ENABLED': 'off'})
        self.assertFalse(config.rate_limit_enabled)

    def test_rate_limit_enabled_empty_string_keeps_default_enabled(self):
        # empty string → _pick returns '' → _parse_bool returns default=True
        config = self._config_from_env({'FMS_RATE_LIMIT_ENABLED': ''})
        self.assertTrue(config.rate_limit_enabled)

    def test_rate_limit_enabled_one_enables(self):
        config = self._config_from_env({'FMS_RATE_LIMIT_ENABLED': '1'})
        self.assertTrue(config.rate_limit_enabled)

    # --- dotenv inline comment stripping ---

    def test_dotenv_inline_comment_stripped_from_int_value(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = os.path.join(tmp, '.env')
            with open(path, 'w') as f:
                f.write('FMS_API_BASE=http://x.test\n')
                f.write('FMS_RATE_LIMIT_MAX_REQUESTS=45 # requests per window\n')
            config = GatewayConfig.from_env(env={}, dotenv_path=path)
            self.assertEqual(45, config.rate_limit_max_requests)

    def test_dotenv_inline_comment_stripped_from_bool_value(self):
        with tempfile.TemporaryDirectory() as tmp:
            path = os.path.join(tmp, '.env')
            with open(path, 'w') as f:
                f.write('FMS_API_BASE=http://x.test\n')
                f.write('FMS_RATE_LIMIT_ENABLED=0 # disabled for testing\n')
            config = GatewayConfig.from_env(env={}, dotenv_path=path)
            self.assertFalse(config.rate_limit_enabled)


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