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

"""Tests documenting critical behavioral semantics of thread_context.py.

These tests document non-obvious behaviors that are important for correctness
when using ThreadPoolExecutor with research context propagation.
"""

from local_deep_research.utilities.thread_context import (
    clear_search_context,
    get_search_context,
    preserve_research_context,
    search_context,
    set_search_context,
)


class TestPreserveResearchContextDecorator:
    """Tests for @preserve_research_context decorator semantics."""

    def test_preserve_research_context_captures_at_decoration_time(self):
        """Context is captured when @preserve_research_context is applied,
        NOT when the decorated function is called. This is the most important
        behavioral property — it's what makes thread context propagation work."""
        # Set context BEFORE decorating
        set_search_context({"research_id": "captured-at-decoration"})

        @preserve_research_context
        def task():
            return get_search_context()

        # Clear context AFTER decorating but BEFORE calling
        clear_search_context()

        # The decorated function should still see the context from decoration time
        result = task()
        assert result is not None
        assert result["research_id"] == "captured-at-decoration"

    def test_multiple_decorated_functions_carry_independent_snapshots(self):
        """Two functions decorated at different times carry their own
        independent context snapshots."""
        set_search_context({"research_id": "snapshot-1", "phase": "first"})

        @preserve_research_context
        def task_1():
            return get_search_context()

        # Change context before decorating second function
        set_search_context({"research_id": "snapshot-2", "phase": "second"})

        @preserve_research_context
        def task_2():
            return get_search_context()

        clear_search_context()

        result_1 = task_1()
        result_2 = task_2()

        assert result_1["research_id"] == "snapshot-1"
        assert result_1["phase"] == "first"
        assert result_2["research_id"] == "snapshot-2"
        assert result_2["phase"] == "second"

    def test_decorator_with_no_context_is_noop(self):
        """When no context is set at decoration time, the decorator
        doesn't fail — it simply doesn't set/clear any context."""
        clear_search_context()

        @preserve_research_context
        def task():
            return get_search_context()

        result = task()
        # No context was set at decoration time, so context remains None
        assert result is None

    def test_decorator_clears_context_in_finally(self):
        """The decorator clears context after the function runs,
        even if the context was set. This prevents leakage in thread pools."""
        set_search_context({"research_id": "will-be-cleared"})

        @preserve_research_context
        def task():
            ctx = get_search_context()
            return ctx

        clear_search_context()

        # Run the decorated function
        task()

        # After the decorated function returns, context should be cleared
        # because the finally block calls clear_search_context()
        assert get_search_context() is None


class TestSearchContextManager:
    """Tests for the search_context context manager."""

    def test_context_manager_destroys_pre_existing_context(self):
        """with search_context(...) clears any previously-set context in finally.
        This is destructive — the pre-existing context is NOT restored."""
        set_search_context({"research_id": "pre-existing"})

        with search_context({"research_id": "temporary"}):
            ctx = get_search_context()
            assert ctx["research_id"] == "temporary"

        # After exiting, context is CLEARED (not restored to pre-existing)
        assert get_search_context() is None

    def test_context_manager_clears_on_exception(self):
        """Context is cleared even if an exception occurs inside the block."""
        try:
            with search_context({"research_id": "will-fail"}):
                raise ValueError("test error")
        except ValueError:
            pass

        assert get_search_context() is None


class TestShallowCopySemantics:
    """Tests documenting shallow copy behavior of get_search_context."""

    def test_shallow_copy_allows_nested_mutation_leakage(self):
        """get_search_context() returns a shallow copy — top-level keys
        are independent, but nested dict mutations leak to the original."""
        original = {
            "research_id": "test",
            "metadata": {"depth": 1, "tags": ["a"]},
        }
        set_search_context(original)

        retrieved = get_search_context()

        # Top-level mutation doesn't leak
        retrieved["research_id"] = "modified"
        assert get_search_context()["research_id"] == "test"

        # But nested mutation DOES leak (shallow copy)
        retrieved["metadata"]["depth"] = 999
        assert get_search_context()["metadata"]["depth"] == 999

        clear_search_context()