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

"""
Comprehensive coverage tests for url_utils.normalize_url and
thread_context (set/get/clear, search_context manager, preserve_research_context).

This file is intentionally self-contained: every scenario listed in the task
specification is covered with distinct, well-named tests.
"""

import sys
import threading
from concurrent.futures import ThreadPoolExecutor
from types import ModuleType
from unittest.mock import MagicMock, patch

import pytest

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


# ===================================================================
# PART 1 -- normalize_url
# ===================================================================


class TestNormalizeUrlLocalhostDefault:
    """The canonical example: localhost:11434 -> http://localhost:11434."""

    def test_localhost_with_port(self):
        assert normalize_url("localhost:11434") == "http://localhost:11434"

    def test_localhost_bare(self):
        assert normalize_url("localhost") == "http://localhost"

    def test_localhost_with_path_and_port(self):
        assert (
            normalize_url("localhost:11434/api/generate")
            == "http://localhost:11434/api/generate"
        )


class TestNormalizeUrlMalformedScheme:
    """http:example.com (missing //) should be repaired."""

    def test_http_colon_no_slashes(self):
        assert normalize_url("http:example.com") == "http://example.com"

    def test_https_colon_no_slashes(self):
        assert normalize_url("https:example.com") == "https://example.com"

    def test_http_colon_with_port(self):
        assert (
            normalize_url("http:example.com:8080") == "http://example.com:8080"
        )

    def test_https_colon_with_path(self):
        assert (
            normalize_url("https:example.com/api/v1")
            == "https://example.com/api/v1"
        )

    def test_http_colon_localhost(self):
        assert normalize_url("http:localhost:11434") == "http://localhost:11434"


class TestNormalizeUrlAlreadyCorrect:
    """URLs with proper scheme are returned unchanged."""

    def test_http_url_unchanged(self):
        assert (
            normalize_url("http://localhost:11434") == "http://localhost:11434"
        )

    def test_https_url_unchanged(self):
        url = "https://api.openai.com/v1/chat/completions"
        assert normalize_url(url) == url

    def test_http_with_path_query_fragment(self):
        url = "http://example.com/path?q=1#frag"
        assert normalize_url(url) == url

    def test_https_with_userinfo(self):
        url = "https://user:pass@host:9200/path"
        assert normalize_url(url) == url


class TestNormalizeUrlIPv6:
    """IPv6 addresses in brackets."""

    def test_ipv6_loopback(self):
        assert normalize_url("[::1]") == "http://[::1]"

    def test_ipv6_loopback_with_port(self):
        assert normalize_url("[::1]:8080") == "http://[::1]:8080"

    def test_ipv6_link_local(self):
        result = normalize_url("[fe80::1]:8080")
        assert result == "http://[fe80::1]:8080"

    def test_ipv6_private_fc00(self):
        result = normalize_url("[fc00::1]:9200")
        assert result == "http://[fc00::1]:9200"

    def test_ipv6_public(self):
        result = normalize_url("[2001:4860:4860::8888]")
        assert result.startswith("https://")

    def test_ipv6_already_has_scheme(self):
        url = "http://[::1]:11434"
        assert normalize_url(url) == url


class TestNormalizeUrlPrivateIPs:
    """Private IPs get http://."""

    @pytest.mark.parametrize(
        "ip",
        [
            "10.0.0.1",
            "10.255.255.255",
            "172.16.0.1",
            "172.31.255.255",
            "192.168.0.1",
            "192.168.1.100",
        ],
    )
    def test_rfc1918_gets_http(self, ip):
        assert normalize_url(ip) == f"http://{ip}"

    def test_private_ip_with_port(self):
        assert (
            normalize_url("192.168.1.100:8888") == "http://192.168.1.100:8888"
        )

    def test_zero_address(self):
        assert normalize_url("0.0.0.0") == "http://0.0.0.0"

    def test_127_0_0_1(self):
        assert normalize_url("127.0.0.1") == "http://127.0.0.1"

    def test_link_local_169_254(self):
        assert normalize_url("169.254.0.1") == "http://169.254.0.1"


class TestNormalizeUrlExternalHosts:
    """External / public hosts get https://."""

    @pytest.mark.parametrize(
        "host",
        [
            "8.8.8.8",
            "1.1.1.1",
            "example.com",
            "api.openai.com",
            "sub.domain.example.org",
        ],
    )
    def test_public_host_gets_https(self, host):
        assert normalize_url(host) == f"https://{host}"

    def test_public_ip_with_port(self):
        assert normalize_url("8.8.8.8:53") == "https://8.8.8.8:53"


class TestNormalizeUrlEmptyAndWhitespace:
    """Empty string and whitespace edge cases."""

    def test_empty_string_raises(self):
        with pytest.raises(ValueError, match="(?i)empty"):
            normalize_url("")

    def test_whitespace_stripped_leading_trailing(self):
        assert normalize_url("  https://example.com  ") == "https://example.com"

    def test_whitespace_around_bare_host(self):
        assert normalize_url("  localhost:8080  ") == "http://localhost:8080"

    def test_tab_stripped(self):
        assert normalize_url("\thttps://example.com\t") == "https://example.com"

    def test_whitespace_only_does_not_raise(self):
        # " " is truthy so it passes the empty check, but strips to ""
        # which falls through to the scheme-adding logic
        result = normalize_url("   ")
        assert isinstance(result, str)


class TestNormalizeUrlDoubleSlash:
    """Protocol-relative //hostname format."""

    def test_double_slash_public(self):
        assert normalize_url("//example.com") == "https://example.com"

    def test_double_slash_public_with_port(self):
        assert normalize_url("//example.com:443") == "https://example.com:443"

    def test_double_slash_localhost(self):
        assert normalize_url("//localhost:8080") == "http://localhost:8080"

    def test_double_slash_private_ip(self):
        assert normalize_url("//192.168.1.1:9200") == "http://192.168.1.1:9200"

    def test_double_slash_with_path(self):
        assert (
            normalize_url("//example.com/api/v1")
            == "https://example.com/api/v1"
        )


class TestNormalizeUrlMockedPrivateIp:
    """Mock-based tests to force both branches of the private-ip decision."""

    def test_mock_private_gives_http(self):
        with patch(
            "local_deep_research.utilities.url_utils.is_private_ip",
            return_value=True,
        ):
            assert normalize_url("anyhost:1234") == "http://anyhost:1234"

    def test_mock_public_gives_https(self):
        with patch(
            "local_deep_research.utilities.url_utils.is_private_ip",
            return_value=False,
        ):
            assert normalize_url("anyhost:1234") == "https://anyhost:1234"

    def test_ipv6_hostname_passed_with_brackets(self):
        """Bracketed IPv6 passes bracket-inclusive hostname to is_private_ip."""
        with patch(
            "local_deep_research.utilities.url_utils.is_private_ip",
            return_value=True,
        ) as mock_fn:
            normalize_url("[::1]:8080")
            mock_fn.assert_called_once_with("[::1]")

    def test_bare_hostname_extracted_before_colon(self):
        with patch(
            "local_deep_research.utilities.url_utils.is_private_ip",
            return_value=False,
        ) as mock_fn:
            normalize_url("myhost:9999")
            mock_fn.assert_called_once_with("myhost")

    def test_bare_hostname_extracted_before_slash(self):
        with patch(
            "local_deep_research.utilities.url_utils.is_private_ip",
            return_value=False,
        ) as mock_fn:
            normalize_url("myhost/some/path")
            mock_fn.assert_called_once_with("myhost")


# ===================================================================
# PART 2 -- thread_context
# ===================================================================


class TestSetSearchContext:
    """set_search_context sets context and overwrites previous."""

    def setup_method(self):
        clear_search_context()

    def teardown_method(self):
        clear_search_context()

    def test_sets_context(self):
        set_search_context({"research_id": "abc"})
        assert get_search_context() == {"research_id": "abc"}

    def test_overwrites_previous(self):
        set_search_context({"first": 1})
        set_search_context({"second": 2})
        assert get_search_context() == {"second": 2}

    def test_makes_defensive_copy(self):
        original = {"key": "value"}
        set_search_context(original)
        original["key"] = "mutated"
        assert get_search_context()["key"] == "value"

    def test_empty_dict_accepted(self):
        set_search_context({})
        assert get_search_context() == {}


class TestClearSearchContext:
    """clear_search_context removes the context attribute entirely."""

    def setup_method(self):
        clear_search_context()

    def teardown_method(self):
        clear_search_context()

    def test_clears_existing(self):
        set_search_context({"x": 1})
        clear_search_context()
        assert get_search_context() is None

    def test_noop_when_nothing_set(self):
        clear_search_context()  # must not raise

    def test_clear_then_set_again(self):
        set_search_context({"a": 1})
        clear_search_context()
        set_search_context({"b": 2})
        assert get_search_context() == {"b": 2}


class TestGetSearchContext:
    """get_search_context returns a copy, or None when unset."""

    def setup_method(self):
        clear_search_context()

    def teardown_method(self):
        clear_search_context()

    def test_returns_none_when_not_set(self):
        assert get_search_context() is None

    def test_returns_copy_not_reference(self):
        set_search_context({"key": "value"})
        result = get_search_context()
        result["key"] = "mutated"
        assert get_search_context()["key"] == "value"

    def test_two_gets_are_independent_copies(self):
        set_search_context({"k": "v"})
        a = get_search_context()
        b = get_search_context()
        assert a == b
        assert a is not b

    def test_returns_correct_data(self):
        set_search_context({"research_id": "r1", "extra": 42})
        ctx = get_search_context()
        assert ctx["research_id"] == "r1"
        assert ctx["extra"] == 42


class TestSearchContextManager:
    """search_context context manager sets on entry and clears on exit."""

    def setup_method(self):
        clear_search_context()

    def teardown_method(self):
        clear_search_context()

    def test_sets_inside_block(self):
        with search_context({"research_id": "cm"}):
            assert get_search_context() == {"research_id": "cm"}

    def test_clears_after_block(self):
        with search_context({"research_id": "cm2"}):
            pass
        assert get_search_context() is None

    def test_clears_on_exception(self):
        with pytest.raises(RuntimeError):
            with search_context({"research_id": "err"}):
                raise RuntimeError("boom")
        assert get_search_context() is None

    def test_copies_input_dict(self):
        original = {"key": "val"}
        with search_context(original):
            original["key"] = "changed"
            assert get_search_context()["key"] == "val"

    def test_nested_inner_clears(self):
        """Inner context manager clears on exit (no stack)."""
        with search_context({"level": "outer"}):
            with search_context({"level": "inner"}):
                assert get_search_context()["level"] == "inner"
            # Inner exited, context is now gone
            assert get_search_context() is None


class TestPreserveResearchContext:
    """preserve_research_context decorator captures and restores context."""

    def setup_method(self):
        clear_search_context()
        # Evict the module so each test exercises a fresh lazy import; keep
        # the original so teardown can put it back instead of leaving the
        # NEXT importer to build a new singleton distinct from the one
        # earlier-bound consumers hold.
        self._saved_tls_module = sys.modules.pop(
            "local_deep_research.database.thread_local_session", None
        )

    def teardown_method(self):
        clear_search_context()
        sys.modules.pop(
            "local_deep_research.database.thread_local_session", None
        )
        db_pkg = sys.modules.get("local_deep_research.database")
        if self._saved_tls_module is not None:
            sys.modules["local_deep_research.database.thread_local_session"] = (
                self._saved_tls_module
            )
            if db_pkg is not None:
                db_pkg.thread_local_session = self._saved_tls_module
        elif db_pkg is not None and hasattr(db_pkg, "thread_local_session"):
            # The test's fresh import bound a throwaway module on the parent
            # package; drop it so attribute lookups can't resolve to it.
            delattr(db_pkg, "thread_local_session")

    def test_captures_context_at_decoration_time(self):
        set_search_context({"research_id": "snap"})

        @preserve_research_context
        def worker():
            return get_search_context()

        # Clear the context on the main thread after decoration
        clear_search_context()

        # The decorator captured the snapshot; running in another thread
        # should still see it.
        with ThreadPoolExecutor(max_workers=1) as pool:
            result = pool.submit(worker).result()

        assert result == {"research_id": "snap"}

    def test_restores_and_clears_in_worker_thread(self):
        set_search_context({"research_id": "restore"})

        context_after = []

        @preserve_research_context
        def worker():
            return "done"

        def run_then_check():
            worker()
            context_after.append(get_search_context())

        clear_search_context()

        with ThreadPoolExecutor(max_workers=1) as pool:
            pool.submit(run_then_check).result()

        # Context should have been cleared after the worker finished
        assert context_after[0] is None

    def test_no_context_runs_without_error(self):
        """When no context was set at decoration time, runs normally."""
        assert get_search_context() is None

        @preserve_research_context
        def worker():
            return "ok"

        assert worker() == "ok"

    def test_preserves_function_name(self):
        @preserve_research_context
        def my_func():
            """Docstring."""
            pass

        assert my_func.__name__ == "my_func"
        assert my_func.__doc__ == "Docstring."

    def test_passes_args_and_kwargs(self):
        set_search_context({"research_id": "args"})

        @preserve_research_context
        def add(a, b, offset=0):
            return a + b + offset

        clear_search_context()

        with ThreadPoolExecutor(max_workers=1) as pool:
            result = pool.submit(add, 3, 4, offset=10).result()

        assert result == 17

    def test_exception_propagates(self):
        set_search_context({"research_id": "exc"})

        @preserve_research_context
        def boom():
            raise ValueError("kaboom")

        clear_search_context()

        with pytest.raises(ValueError, match="kaboom"):
            boom()

    def test_clears_context_even_on_exception(self):
        set_search_context({"research_id": "exc2"})

        context_after = []

        @preserve_research_context
        def boom():
            raise ValueError("kaboom")

        clear_search_context()

        def run_then_check():
            try:
                boom()
            except ValueError:
                pass
            context_after.append(get_search_context())

        with ThreadPoolExecutor(max_workers=1) as pool:
            pool.submit(run_then_check).result()

        assert context_after[0] is None

    def test_cleanup_current_thread_called(self):
        """The finally block should call cleanup_current_thread."""
        set_search_context({"research_id": "cleanup"})

        @preserve_research_context
        def worker():
            return 42

        clear_search_context()

        mock_cleanup = MagicMock()
        fake_module = ModuleType(
            "local_deep_research.database.thread_local_session"
        )
        fake_module.cleanup_current_thread = mock_cleanup

        with patch.dict(
            sys.modules,
            {
                "local_deep_research.database.thread_local_session": (
                    fake_module
                )
            },
        ):
            result = worker()

        assert result == 42
        mock_cleanup.assert_called_once()

    def test_cleanup_exception_is_suppressed(self):
        """If cleanup_current_thread raises, the error is swallowed."""
        set_search_context({"research_id": "suppress"})

        @preserve_research_context
        def worker():
            return "fine"

        clear_search_context()

        fake_module = ModuleType(
            "local_deep_research.database.thread_local_session"
        )
        fake_module.cleanup_current_thread = MagicMock(
            side_effect=RuntimeError("db engine died")
        )

        with patch.dict(
            sys.modules,
            {
                "local_deep_research.database.thread_local_session": (
                    fake_module
                )
            },
        ):
            result = worker()

        assert result == "fine"

    def test_cleanup_import_failure_suppressed(self):
        """If thread_local_session module cannot be imported, no crash."""
        set_search_context({"research_id": "noimport"})

        @preserve_research_context
        def worker():
            return "still fine"

        clear_search_context()

        with patch.dict(
            sys.modules,
            {"local_deep_research.database.thread_local_session": None},
        ):
            assert worker() == "still fine"


class TestThreadIsolation:
    """Context is truly thread-local: other threads cannot see it."""

    def setup_method(self):
        clear_search_context()

    def teardown_method(self):
        clear_search_context()

    def test_other_thread_sees_none(self):
        set_search_context({"main": True})

        result = []

        def check():
            result.append(get_search_context())

        t = threading.Thread(target=check)
        t.start()
        t.join()

        assert result[0] is None

    def test_threads_have_independent_contexts(self):
        results = {}

        def worker(tid):
            set_search_context({"tid": tid})
            ctx = get_search_context()
            results[tid] = ctx["tid"]

        threads = [threading.Thread(target=worker, args=(i,)) for i in range(5)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()

        for i in range(5):
            assert results[i] == i