"""Tests for threading_utils module.""" from concurrent.futures import ThreadPoolExecutor from unittest.mock import MagicMock, Mock, patch import pytest from local_deep_research.utilities.threading_utils import ( g_thread_local_store, thread_context, thread_specific_cache, thread_with_app_context, ) class TestThreadSpecificCache: """Tests for thread_specific_cache decorator.""" def setup_method(self): """Clear thread-local data before each test.""" if hasattr(g_thread_local_store, "thread_id"): delattr(g_thread_local_store, "thread_id") def test_caches_result_in_same_thread(self): """Should cache results within the same thread.""" call_count = 0 @thread_specific_cache({}) def expensive_func(x): nonlocal call_count call_count += 1 return x * 2 # First call result1 = expensive_func(5) # Second call with same arg result2 = expensive_func(5) assert result1 == 10 assert result2 == 10 assert call_count == 1 # Should only be called once def test_different_args_not_cached(self): """Different arguments should not use cached result.""" call_count = 0 @thread_specific_cache({}) def func(x): nonlocal call_count call_count += 1 return x * 2 result1 = func(5) result2 = func(10) assert result1 == 10 assert result2 == 20 assert call_count == 2 def test_cache_isolated_between_threads(self): """Cache should be isolated between threads.""" call_counts = {"main": 0, "other": 0} @thread_specific_cache({}) def func(thread_name, x): call_counts[thread_name] += 1 return x * 2 # Call in main thread result_main = func("main", 5) # Call in other thread with same args def other_thread(): return func("other", 5) with ThreadPoolExecutor(max_workers=1) as executor: future = executor.submit(other_thread) result_other = future.result() assert result_main == 10 assert result_other == 10 # Both threads should have called the function assert call_counts["main"] == 1 assert call_counts["other"] == 1 def test_assigns_unique_thread_id(self): """Should assign unique thread ID to each thread.""" thread_ids = [] @thread_specific_cache({}) def func(): thread_ids.append(g_thread_local_store.thread_id) return "done" func() main_id = g_thread_local_store.thread_id def other_thread(): func() return g_thread_local_store.thread_id with ThreadPoolExecutor(max_workers=1) as executor: future = executor.submit(other_thread) other_id = future.result() assert main_id != other_id def test_reuses_thread_id_within_thread(self): """Should reuse same thread ID for multiple calls in same thread.""" @thread_specific_cache({}) def func(x): return x func(1) id1 = g_thread_local_store.thread_id func(2) id2 = g_thread_local_store.thread_id assert id1 == id2 def test_handles_kwargs(self): """Should properly handle keyword arguments.""" call_count = 0 @thread_specific_cache({}) def func(a, b=None): nonlocal call_count call_count += 1 return (a, b) result1 = func(1, b=2) result2 = func(1, b=2) assert result1 == (1, 2) assert result2 == (1, 2) assert call_count == 1 class TestThreadWithAppContext: """Tests for thread_with_app_context decorator.""" def test_runs_with_app_context(self): """Should run function within provided app context.""" mock_context = MagicMock() mock_context.__enter__ = Mock(return_value=None) mock_context.__exit__ = Mock(return_value=None) @thread_with_app_context def my_func(x): return x * 2 result = my_func(mock_context, 5) assert result == 10 mock_context.__enter__.assert_called_once() mock_context.__exit__.assert_called_once() def test_runs_without_context_when_none(self): """Should run function normally when context is None.""" @thread_with_app_context def my_func(x): return x * 3 result = my_func(None, 7) assert result == 21 def test_preserves_function_metadata(self): """Should preserve function name and docstring.""" @thread_with_app_context def documented_func(): """My documentation.""" pass assert documented_func.__name__ == "documented_func" assert documented_func.__doc__ == "My documentation." def test_passes_args_correctly(self): """Should pass arguments correctly to wrapped function.""" mock_context = MagicMock() mock_context.__enter__ = Mock(return_value=None) mock_context.__exit__ = Mock(return_value=None) @thread_with_app_context def func(a, b, c=None): return (a, b, c) result = func(mock_context, "x", "y", c="z") assert result == ("x", "y", "z") def test_context_manager_properly_exited_on_exception(self): """Context manager should be properly exited even on exception.""" mock_context = MagicMock() mock_context.__enter__ = Mock(return_value=None) mock_context.__exit__ = Mock(return_value=False) @thread_with_app_context def failing_func(): raise ValueError("Test error") with pytest.raises(ValueError, match="Test error"): failing_func(mock_context) mock_context.__exit__.assert_called_once() class TestThreadContext: """Tests for thread_context function.""" def test_returns_none_without_flask_context(self): """Should return None when no Flask app context is active.""" import local_deep_research.utilities.threading_utils as module mock_app = Mock() mock_app.app_context.side_effect = RuntimeError("No context") with patch.object(module, "current_app", mock_app): with patch.object(module, "g", Mock()): result = thread_context() assert result is None def test_returns_app_context_when_available(self): """Should return app context when Flask context is active.""" import local_deep_research.utilities.threading_utils as module mock_context = MagicMock() mock_context.__enter__ = Mock(return_value=None) mock_context.__exit__ = Mock(return_value=None) mock_app = Mock() mock_app.app_context.return_value = mock_context mock_g = Mock() mock_g.__iter__ = Mock(return_value=iter([])) with patch.object(module, "current_app", mock_app): with patch.object(module, "g", mock_g): result = thread_context() assert result == mock_context def test_copies_global_data(self): """Should copy global data from current context.""" import local_deep_research.utilities.threading_utils as module mock_context = MagicMock() mock_context.__enter__ = Mock(return_value=None) mock_context.__exit__ = Mock(return_value=None) mock_app = Mock() mock_app.app_context.return_value = mock_context mock_g = Mock() mock_g.__iter__ = Mock(return_value=iter(["user_id", "settings"])) mock_g.get = Mock( side_effect=lambda k: {"user_id": 123, "settings": {"a": 1}}[k] ) with patch.object(module, "current_app", mock_app): with patch.object(module, "g", mock_g): thread_context() # Should have iterated over g mock_g.__iter__.assert_called() def test_handles_type_error_on_g_iteration(self): """Should handle TypeError when g is not iterable.""" import local_deep_research.utilities.threading_utils as module mock_context = MagicMock() mock_context.__enter__ = Mock(return_value=None) mock_context.__exit__ = Mock(return_value=None) mock_app = Mock() mock_app.app_context.return_value = mock_context mock_g = Mock() mock_g.__iter__ = Mock(side_effect=TypeError("Not iterable")) with patch.object(module, "current_app", mock_app): with patch.object(module, "g", mock_g): # Should not raise, should return context result = thread_context() assert result == mock_context def test_logs_debug_when_no_context(self): """Should log debug message when no app context available.""" import local_deep_research.utilities.threading_utils as module mock_app = Mock() mock_app.app_context.side_effect = RuntimeError("No context") with patch.object(module, "current_app", mock_app): with patch.object(module, "g", Mock()): with patch.object(module, "logger") as mock_logger: thread_context() mock_logger.debug.assert_called() class TestIntegration: """Integration tests for threading utilities.""" def test_thread_specific_cache_with_thread_context(self): """Should work correctly with thread context propagation.""" call_count = 0 @thread_specific_cache({}) def cached_func(x): nonlocal call_count call_count += 1 return x * 2 # Main thread calls cached_func(5) cached_func(5) # Should use cache results = [] def worker(): # Worker thread should have its own cache return cached_func(5) with ThreadPoolExecutor(max_workers=2) as executor: futures = [executor.submit(worker) for _ in range(3)] results = [f.result() for f in futures] # All results should be correct assert all(r == 10 for r in results) # Main thread called once, each worker thread called once # (some workers may share thread due to pool reuse) assert call_count >= 2 # At minimum: main + 1 worker def test_thread_with_app_context_in_thread_pool(self): """Should properly inject app context in thread pool.""" results = [] @thread_with_app_context def worker(value): return value * 2 mock_context = MagicMock() mock_context.__enter__ = Mock(return_value=None) mock_context.__exit__ = Mock(return_value=None) with ThreadPoolExecutor(max_workers=2) as executor: futures = [ executor.submit(worker, mock_context, i) for i in range(5) ] results = [f.result() for f in futures] assert results == [0, 2, 4, 6, 8] class TestEdgeCases: """Edge case tests for threading utilities.""" def setup_method(self): """Clear thread-local data before each test.""" if hasattr(g_thread_local_store, "thread_id"): delattr(g_thread_local_store, "thread_id") def test_thread_specific_cache_with_unhashable_args(self): """Should raise error for unhashable arguments.""" @thread_specific_cache({}) def func(x): return x # Lists are unhashable with pytest.raises(TypeError): func([1, 2, 3]) def test_thread_specific_cache_with_none_arg(self): """Should handle None arguments correctly.""" call_count = 0 @thread_specific_cache({}) def func(x): nonlocal call_count call_count += 1 return x is None result1 = func(None) result2 = func(None) assert result1 is True assert result2 is True assert call_count == 1 def test_nested_decorators(self): """Should work with other decorators.""" from functools import wraps def logging_decorator(func): @wraps(func) def wrapper(*args, **kwargs): return func(*args, **kwargs) return wrapper call_count = 0 @logging_decorator @thread_specific_cache({}) def func(x): nonlocal call_count call_count += 1 return x result1 = func(5) result2 = func(5) assert result1 == 5 assert result2 == 5 assert call_count == 1