Spaces:
Running
Running
| """ | |
| 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}" | |
| ) | |