learningcircuit--local-deep-research/tests/utilities/test_threading_utils.py

"""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