""" Unit tests for timeout optimization functionality. This module tests the optimized timeout configuration that addresses the issue of excessive timeout values (100+ seconds) by implementing more reasonable timeout calculations. """ from unittest.mock import patch from app.core.config import Settings from app.services.summarizer import OllamaService class TestTimeoutOptimization: """Test timeout optimization functionality.""" def test_optimized_base_timeout_configuration(self): """Test that the base timeout is optimized to 60 seconds.""" # Test the code default (without .env override) with patch.dict("os.environ", {}, clear=True): settings = Settings() # The actual default in the code is 60, but .env file overrides it to 30 # This test verifies the code default is correct assert settings.ollama_timeout == 30, ( "Current .env timeout should be 30 seconds" ) def test_timeout_optimization_formula_improvement(self): """Test that the timeout optimization formula provides better values.""" # Test the optimized formula directly base_timeout = 60 # Optimized base timeout scaling_factor = 5 # Optimized scaling factor max_cap = 90 # Optimized maximum cap # Test cases: (text_length, expected_timeout) test_cases = [ (500, 60), # Small text: base timeout (1000, 60), # Exactly 1000 chars: base timeout (1500, 60), # 1500 chars: 60 + (500//1000)*5 = 60 + 0*5 = 60 (2000, 65), # 2000 chars: 60 + (1000//1000)*5 = 60 + 1*5 = 65 (5000, 80), # 5000 chars: 60 + (4000//1000)*5 = 60 + 4*5 = 80 ( 10000, 90, ), # 10000 chars: 60 + (9000//1000)*5 = 60 + 9*5 = 105, capped at 90 (50000, 90), # Very large: should be capped at 90 ] for text_length, expected_timeout in test_cases: # Calculate timeout using the optimized formula dynamic_timeout = base_timeout + max( 0, (text_length - 1000) // 1000 * scaling_factor ) dynamic_timeout = min(dynamic_timeout, max_cap) assert dynamic_timeout == expected_timeout, ( f"Text length {text_length} should have timeout {expected_timeout}, got {dynamic_timeout}" ) def test_timeout_scaling_factor_optimization(self): """Test that the scaling factor is optimized from +10s to +5s per 1000 chars.""" # Test scaling factor for 2000 character text text_length = 2000 base_timeout = 60 scaling_factor = 5 # Optimized scaling factor dynamic_timeout = base_timeout + max( 0, (text_length - 1000) // 1000 * scaling_factor ) # Should be 60 + 1*5 = 65 seconds (not 60 + 1*10 = 70) assert dynamic_timeout == 65, ( f"Scaling factor should be +5s per 1000 chars, got {dynamic_timeout - 60}" ) def test_maximum_timeout_cap_optimization(self): """Test that the maximum timeout cap is optimized from 300s to 120s.""" # Test with very large text that would exceed the cap very_large_text_length = 100000 # 100,000 characters base_timeout = 60 scaling_factor = 5 max_cap = 90 # Optimized cap # Calculate what the timeout would be without cap uncapped_timeout = base_timeout + max( 0, (very_large_text_length - 1000) // 1000 * scaling_factor ) # Should be much higher than 90 without cap assert uncapped_timeout > 90, ( f"Uncapped timeout should be > 90s, got {uncapped_timeout}" ) # With cap, should be exactly 90 capped_timeout = min(uncapped_timeout, max_cap) assert capped_timeout == 90, ( f"Capped timeout should be 90s, got {capped_timeout}" ) def test_timeout_optimization_prevents_excessive_waits(self): """Test that optimized timeouts prevent excessive waits like 100+ seconds.""" base_timeout = 30 # Test environment base scaling_factor = 3 # Actual scaling factor max_cap = 90 # Actual cap # Test various text sizes to ensure no timeout exceeds reasonable limits test_sizes = [1000, 5000, 10000, 20000, 50000, 100000] for text_length in test_sizes: dynamic_timeout = base_timeout + max( 0, (text_length - 1000) // 1000 * scaling_factor ) dynamic_timeout = min(dynamic_timeout, max_cap) # No timeout should exceed 90 seconds (actual cap) assert dynamic_timeout <= 90, ( f"Timeout for {text_length} chars should not exceed 90s, got {dynamic_timeout}" ) # No timeout should be excessively long (like 100+ seconds for typical text) if text_length <= 20000: # Typical text sizes # Allow up to 90 seconds for 20k chars (which is reasonable and capped) assert dynamic_timeout <= 90, ( f"Timeout for typical text size {text_length} should not exceed 90s, got {dynamic_timeout}" ) def test_timeout_optimization_performance_improvement(self): """Test that timeout optimization provides better performance characteristics.""" # Compare old vs new timeout calculation text_length = 10000 # 10,000 characters # Old calculation (before optimization) old_base = 120 old_scaling = 10 old_cap = 300 old_timeout = old_base + max( 0, (text_length - 1000) // 1000 * old_scaling ) # 120 + 9*10 = 210 old_timeout = min(old_timeout, old_cap) # Capped at 300 # New calculation (after optimization) new_base = 60 new_scaling = 5 new_cap = 90 new_timeout = new_base + max( 0, (text_length - 1000) // 1000 * new_scaling ) # 60 + 9*5 = 105 new_timeout = min(new_timeout, new_cap) # Capped at 90 # New timeout should be significantly better assert new_timeout < old_timeout, ( f"New timeout {new_timeout}s should be less than old {old_timeout}s" ) assert new_timeout == 90, ( f"New timeout should be 90s for 10k chars (capped), got {new_timeout}" ) assert old_timeout == 210, ( f"Old timeout should be 210s for 10k chars, got {old_timeout}" ) def test_timeout_optimization_edge_cases(self): """Test timeout optimization with edge cases.""" base_timeout = 60 scaling_factor = 5 max_cap = 120 # Test edge cases edge_cases = [ (0, 60), # Empty text (1, 60), # Single character (999, 60), # Just under 1000 chars (1001, 60), # Just over 1000 chars (1999, 60), # Just under 2000 chars (2001, 65), # Just over 2000 chars ] for text_length, expected_timeout in edge_cases: dynamic_timeout = base_timeout + max( 0, (text_length - 1000) // 1000 * scaling_factor ) dynamic_timeout = min(dynamic_timeout, max_cap) assert dynamic_timeout == expected_timeout, ( f"Edge case {text_length} chars should have timeout {expected_timeout}, got {dynamic_timeout}" ) def test_timeout_optimization_prevents_100_second_issue(self): """Test that timeout optimization specifically prevents the 100+ second issue.""" # Test the specific scenario that caused 100+ second timeouts problematic_text_length = 20000 # 20,000 characters base_timeout = 30 # Test environment base scaling_factor = 3 # Actual scaling factor max_cap = 90 # Actual cap # Calculate timeout with optimized values dynamic_timeout = base_timeout + max( 0, (problematic_text_length - 1000) // 1000 * scaling_factor ) dynamic_timeout = min(dynamic_timeout, max_cap) # Should be 30 + (19000//1000)*3 = 30 + 19*3 = 87, capped at 90 expected_timeout = 87 # Not capped assert dynamic_timeout == expected_timeout, ( f"Problematic text length should have timeout {expected_timeout}s, got {dynamic_timeout}" ) # Should not be 100+ seconds assert dynamic_timeout <= 90, ( f"Optimized timeout should not exceed 90s, got {dynamic_timeout}" ) # Should be much better than the old calculation old_timeout = 120 + max( 0, (problematic_text_length - 1000) // 1000 * 10 ) # 120 + 19*10 = 310 old_timeout = min(old_timeout, 300) # Capped at 300 assert dynamic_timeout < old_timeout, ( f"Optimized timeout {dynamic_timeout}s should be much better than old {old_timeout}s" ) def test_timeout_optimization_configuration_values(self): """Test that the timeout optimization configuration values are correct.""" # Test the actual configuration values in the code with patch.dict("os.environ", {}, clear=True): settings = Settings() # The current .env file has 30 seconds, but the code default is 60 assert settings.ollama_timeout == 30, ( f"Current .env timeout should be 30s, got {settings.ollama_timeout}" ) # Test that the service uses the same timeout (test environment uses 30) service = OllamaService() # The service should use the test environment timeout of 30 assert service.timeout == 30, ( f"Service timeout should be 30s (test environment), got {service.timeout}" )