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_diagnostic_reporter_config_reads_all_operational_limits(self): config = GatewayConfig.from_env(env={ 'FMS_API_BASE': 'https://base.example.com', 'MCP_DIAGNOSIS_ENABLED': 'true', 'MCP_DIAGNOSIS_URL': 'https://support.internal/internal/mcp-diagnostics/events', 'MCP_DIAGNOSIS_KEY_ID': 'gateway-current', 'MCP_DIAGNOSIS_SECRET': 's' * 32, 'MCP_DIAGNOSIS_QUEUE_SIZE': '500', 'MCP_DIAGNOSIS_BATCH_SIZE': '50', 'MCP_DIAGNOSIS_TIMEOUT_SECONDS': '0.4', 'MCP_DIAGNOSIS_INITIAL_BACKOFF_SECONDS': '0.2', 'MCP_DIAGNOSIS_MAX_BACKOFF_SECONDS': '3.0', 'MCP_DIAGNOSIS_ALLOW_INSECURE_HTTP': 'true', }, dotenv_path='missing.env') self.assertTrue(config.diagnosis_enabled) self.assertEqual( 'https://support.internal/internal/mcp-diagnostics/events', config.diagnosis_url, ) self.assertEqual('gateway-current', config.diagnosis_key_id) self.assertEqual('s' * 32, config.diagnosis_secret) self.assertEqual(500, config.diagnosis_queue_size) self.assertEqual(50, config.diagnosis_batch_size) self.assertEqual(0.4, config.diagnosis_timeout_seconds) self.assertEqual(0.2, config.diagnosis_initial_backoff_seconds) self.assertEqual(3.0, config.diagnosis_max_backoff_seconds) self.assertTrue(config.diagnosis_allow_insecure_http) default_config = GatewayConfig.from_env(env={ 'FMS_API_BASE': 'https://base.example.com', }, dotenv_path='missing.env') self.assertFalse(default_config.diagnosis_allow_insecure_http) 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_invalid_diagnostic_float_uses_default(self): config = GatewayConfig.from_env(env={ 'FMS_API_BASE': 'https://base.example.com', 'MCP_DIAGNOSIS_TIMEOUT_SECONDS': 'not-a-number', }, dotenv_path='missing.env') self.assertEqual(0.5, config.diagnosis_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()