feat: add CI timeout protection for tests

pyproject.toml

@@ -152,7 +152,7 @@ markers = [
     "slow: marks tests as slow (deselect with '-m \"not slow\"')",
     "openai: marks tests that require OpenAI API key",
 ]
-timeout = 600
+timeout = 300  # Reduced from 600s (10min) to 300s (5min) for CI safety
 addopts = [
     "-v",
     "--tb=short",

tests/test_basic.py

@@ -7,6 +7,7 @@ import tempfile
 from pathlib import Path
 
 import pytest
+from test_timeout import ci_timeout
 
 
 def test_imports():
@@ -19,6 +20,7 @@ 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
@@ -68,6 +70,7 @@ 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

tests/test_readme_examples.py

@@ -8,9 +8,11 @@ import tempfile
 from pathlib import Path
 
 import pytest
+from test_timeout import ci_timeout
 
 
 @pytest.mark.parametrize("backend_name", ["hnsw", "diskann"])
+@ci_timeout(90)  # 90 second timeout for this comprehensive test
 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
@@ -79,6 +81,7 @@ def test_readme_imports():
     assert callable(LeannChat)
 
 
+@ci_timeout(60)  # 60 second timeout
 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
@@ -115,6 +118,7 @@ def test_backend_options():
 
 
 @pytest.mark.parametrize("backend_name", ["hnsw", "diskann"])
+@ci_timeout(75)  # 75 second timeout for LLM tests
 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

tests/test_timeout.py

@@ -0,0 +1,129 @@
+"""
+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