From f096e62bfa73ce7bda71d5ed4601e42a1f7d28a9 Mon Sep 17 00:00:00 2001 From: Andy Lee Date: Wed, 13 Aug 2025 17:37:43 -0700 Subject: [PATCH] tests: drop custom ci_timeout decorator and helpers; rely on pytest defaults and simplified CI --- tests/test_basic.py | 3 - tests/test_readme_examples.py | 4 -- tests/test_timeout.py | 129 ---------------------------------- 3 files changed, 136 deletions(-) delete mode 100644 tests/test_timeout.py diff --git a/tests/test_basic.py b/tests/test_basic.py index 8ff8caf..800b0ac 100644 --- a/tests/test_basic.py +++ b/tests/test_basic.py @@ -7,7 +7,6 @@ import tempfile from pathlib import Path import pytest -from test_timeout import ci_timeout def test_imports(): @@ -20,7 +19,6 @@ def test_imports(): os.environ.get("CI") == "true", reason="Skip model tests in CI to avoid MPS memory issues" ) @pytest.mark.parametrize("backend_name", ["hnsw", "diskann"]) -@ci_timeout(120) # 2 minute timeout for backend tests def test_backend_basic(backend_name): """Test basic functionality for each backend.""" from leann.api import LeannBuilder, LeannSearcher, SearchResult @@ -70,7 +68,6 @@ def test_backend_basic(backend_name): @pytest.mark.skipif( os.environ.get("CI") == "true", reason="Skip model tests in CI to avoid MPS memory issues" ) -@ci_timeout(180) # 3 minute timeout for large index test def test_large_index(): """Test with larger dataset.""" from leann.api import LeannBuilder, LeannSearcher diff --git a/tests/test_readme_examples.py b/tests/test_readme_examples.py index 4defcf9..fe69e4a 100644 --- a/tests/test_readme_examples.py +++ b/tests/test_readme_examples.py @@ -8,11 +8,9 @@ import tempfile from pathlib import Path import pytest -from test_timeout import ci_timeout @pytest.mark.parametrize("backend_name", ["hnsw", "diskann"]) -@ci_timeout(180) # 180 second timeout to allow for model download and initialization def test_readme_basic_example(backend_name): """Test the basic example from README.md with both backends.""" # Skip on macOS CI due to MPS environment issues with all-MiniLM-L6-v2 @@ -81,7 +79,6 @@ def test_readme_imports(): assert callable(LeannChat) -@ci_timeout(150) # 150 second timeout to allow for model download def test_backend_options(): """Test different backend options mentioned in documentation.""" # Skip on macOS CI due to MPS environment issues with all-MiniLM-L6-v2 @@ -118,7 +115,6 @@ def test_backend_options(): @pytest.mark.parametrize("backend_name", ["hnsw", "diskann"]) -@ci_timeout(150) # 150 second timeout to allow for model download def test_llm_config_simulated(backend_name): """Test simulated LLM configuration option with both backends.""" # Skip on macOS CI due to MPS environment issues with all-MiniLM-L6-v2 diff --git a/tests/test_timeout.py b/tests/test_timeout.py deleted file mode 100644 index d9a64c0..0000000 --- a/tests/test_timeout.py +++ /dev/null @@ -1,129 +0,0 @@ -""" -Test timeout utilities for CI environments. -""" - -import functools -import os -import signal -import sys -from typing import Any, Callable - - -def timeout_test(seconds: int = 30): - """ - Decorator to add timeout to test functions, especially useful in CI environments. - - Args: - seconds: Timeout in seconds (default: 30) - """ - - def decorator(func: Callable) -> Callable: - @functools.wraps(func) - def wrapper(*args: Any, **kwargs: Any) -> Any: - # Only apply timeout in CI environment - if os.environ.get("CI") != "true": - return func(*args, **kwargs) - - # Set up timeout handler - def timeout_handler(signum, frame): - print(f"\n❌ Test {func.__name__} timed out after {seconds} seconds in CI!") - print("This usually indicates a hanging process or infinite loop.") - # Try to cleanup any hanging processes - try: - import subprocess - - subprocess.run( - ["pkill", "-f", "embedding_server"], capture_output=True, timeout=2 - ) - subprocess.run( - ["pkill", "-f", "hnsw_embedding"], capture_output=True, timeout=2 - ) - except Exception: - pass - # Exit with timeout code - sys.exit(124) # Standard timeout exit code - - # Set signal handler and alarm - old_handler = signal.signal(signal.SIGALRM, timeout_handler) - signal.alarm(seconds) - - try: - result = func(*args, **kwargs) - signal.alarm(0) # Cancel alarm - return result - except Exception: - signal.alarm(0) # Cancel alarm on exception - raise - finally: - # Restore original handler - signal.signal(signal.SIGALRM, old_handler) - - return wrapper - - return decorator - - -def ci_timeout(seconds: int = 60): - """ - Timeout decorator specifically for CI environments. - Uses threading for more reliable timeout handling. - - Args: - seconds: Timeout in seconds (default: 60) - """ - - def decorator(func: Callable) -> Callable: - @functools.wraps(func) - def wrapper(*args: Any, **kwargs: Any) -> Any: - # Only apply in CI - if os.environ.get("CI") != "true": - return func(*args, **kwargs) - - import threading - - result = [None] - exception = [None] - finished = threading.Event() - - def target(): - try: - result[0] = func(*args, **kwargs) - except Exception as e: - exception[0] = e - finally: - finished.set() - - # Start function in thread - thread = threading.Thread(target=target, daemon=True) - thread.start() - - # Wait for completion or timeout - if not finished.wait(timeout=seconds): - print(f"\n💥 CI TIMEOUT: Test {func.__name__} exceeded {seconds}s limit!") - print("This usually indicates hanging embedding servers or infinite loops.") - - # Try to cleanup embedding servers - try: - import subprocess - - subprocess.run( - ["pkill", "-9", "-f", "embedding_server"], capture_output=True, timeout=2 - ) - subprocess.run( - ["pkill", "-9", "-f", "hnsw_embedding"], capture_output=True, timeout=2 - ) - print("Attempted to kill hanging embedding servers.") - except Exception as e: - print(f"Cleanup failed: {e}") - - # Raise TimeoutError instead of sys.exit for better pytest integration - raise TimeoutError(f"Test {func.__name__} timed out after {seconds} seconds") - - if exception[0]: - raise exception[0] - - return result[0] - - return wrapper - - return decorator