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