"""Tests for ``_close_base_llm`` covering both sync and async httpx clients.

Background (issue #3816): ``ChatOllama`` owns both ``_client`` (sync
``ollama.Client`` wrapping ``httpx.Client``) and ``_async_client``
(async ``ollama.AsyncClient`` wrapping ``httpx.AsyncClient``). Earlier
versions of ``_close_base_llm`` only closed the sync side, leaking the
async transport per ``ainvoke()`` call — visible as ``a_inode [eventpoll]``
FDs in the issue's lsof dump.
"""

import asyncio
import gc
import os
import resource
import sys
from unittest.mock import Mock

import httpx
import pytest

from local_deep_research.utilities.llm_utils import _close_base_llm


def _open_fd_count() -> int:
    """Test-local file-descriptor counter.

    Inlined here to avoid coupling these tests to a private helper in
    an unrelated production module. On Linux uses ``/proc/self/fd``
    (fast); on macOS falls back to scanning ``RLIMIT_NOFILE``.
    """
    try:
        return len(os.listdir("/proc/self/fd"))
    except (FileNotFoundError, OSError):
        soft_limit = resource.getrlimit(resource.RLIMIT_NOFILE)[0]
        count = 0
        for fd in range(soft_limit):
            try:
                os.fstat(fd)
                count += 1
            except OSError:
                pass
        return count


def _make_fake_chat_ollama(*, sync_close=None, async_aclose=None):
    """Build a stand-in for ``ChatOllama`` with the same private-attr shape.

    The module-string check in ``_close_base_llm`` looks at
    ``type(ollama_client).__module__`` — we set ``__module__`` on the mock's
    type to "ollama" so the introspection treats them as real ollama clients.
    """
    llm = Mock(spec=[])  # spec=[] means hasattr returns False for everything

    if sync_close is not None:
        sync_httpx = Mock(spec=["close"])
        sync_httpx.close = sync_close
        sync_ollama = type("FakeSyncOllama", (), {})()
        sync_ollama._client = sync_httpx
        type(sync_ollama).__module__ = "ollama"
        llm._client = sync_ollama
    else:
        llm._client = None

    if async_aclose is not None:
        async_httpx = Mock(spec=["aclose"])
        async_httpx.aclose = async_aclose
        async_ollama = type("FakeAsyncOllama", (), {})()
        async_ollama._client = async_httpx
        type(async_ollama).__module__ = "ollama"
        llm._async_client = async_ollama
    else:
        llm._async_client = None

    return llm


class TestCloseBaseLLMSync:
    """The sync httpx client must be closed on every call until idempotent."""

    def test_closes_sync_httpx_client(self):
        sync_close = Mock()
        llm = _make_fake_chat_ollama(sync_close=sync_close)

        _close_base_llm(llm)

        sync_close.assert_called_once()

    def test_idempotent_via_ldr_closed_flag(self):
        sync_close = Mock()
        llm = _make_fake_chat_ollama(sync_close=sync_close)

        _close_base_llm(llm)
        _close_base_llm(llm)
        _close_base_llm(llm)

        # Despite three calls, close runs once — _ldr_closed sentinel guards.
        sync_close.assert_called_once()

    def test_swallows_close_exception_and_still_marks_closed(self):
        sync_close = Mock(side_effect=RuntimeError("boom"))
        llm = _make_fake_chat_ollama(sync_close=sync_close)

        # Must not propagate; logs at warning.
        _close_base_llm(llm)

        sync_close.assert_called_once()
        # Subsequent call is skipped by _ldr_closed (no infinite retry).
        _close_base_llm(llm)
        sync_close.assert_called_once()

    def test_handles_missing_async_client_gracefully(self):
        # No _async_client attr at all — must not crash.
        sync_close = Mock()
        llm = _make_fake_chat_ollama(sync_close=sync_close)
        del llm._async_client  # simulate older ollama versions

        _close_base_llm(llm)

        sync_close.assert_called_once()


class TestCloseBaseLLMAsync:
    """The async httpx client must be closed via ``asyncio.run`` when no loop
    is running; skipped (left for the loop owner) otherwise."""

    def test_closes_async_httpx_client_via_asyncio_run(self):
        called = {"count": 0}

        async def fake_aclose():
            called["count"] += 1

        llm = _make_fake_chat_ollama(async_aclose=fake_aclose)

        # No running loop here — _close_base_llm should spin one via
        # asyncio.run() and await aclose().
        _close_base_llm(llm)

        assert called["count"] == 1

    def test_async_close_is_idempotent(self):
        called = {"count": 0}

        async def fake_aclose():
            called["count"] += 1

        llm = _make_fake_chat_ollama(async_aclose=fake_aclose)

        _close_base_llm(llm)
        _close_base_llm(llm)

        # _ldr_closed sentinel prevents the second aclose.
        assert called["count"] == 1

    def test_skips_async_close_inside_running_loop(self):
        called = {"count": 0}

        async def fake_aclose():
            called["count"] += 1

        llm = _make_fake_chat_ollama(async_aclose=fake_aclose)

        async def driver():
            # Inside a running loop, _close_base_llm must NOT call aclose
            # (would deadlock on asyncio.run nesting). The loop owner is
            # responsible for closing the client.
            _close_base_llm(llm)

        asyncio.run(driver())

        # aclose was skipped because we were inside a running loop.
        assert called["count"] == 0
        # And _ldr_closed is NOT set, so a later sync call still tries.
        async_httpx = llm._async_client._client
        assert not getattr(async_httpx, "_ldr_closed", False)

    def test_swallows_async_close_exception(self):
        async def fake_aclose():
            raise RuntimeError("boom")

        llm = _make_fake_chat_ollama(async_aclose=fake_aclose)

        # Must not propagate; logs at warning. _ldr_closed is set so we don't
        # retry endlessly on a known-broken close.
        _close_base_llm(llm)

        async_httpx = llm._async_client._client
        assert getattr(async_httpx, "_ldr_closed", False) is True


class TestCloseBaseLLMBoth:
    """Sync and async sides should both close in the common case."""

    def test_closes_both_sync_and_async(self):
        sync_close = Mock()
        async_called = {"count": 0}

        async def fake_aclose():
            async_called["count"] += 1

        llm = _make_fake_chat_ollama(
            sync_close=sync_close, async_aclose=fake_aclose
        )

        _close_base_llm(llm)

        sync_close.assert_called_once()
        assert async_called["count"] == 1


class TestCloseBaseLLMNonOllama:
    """Non-Ollama LLMs must be left alone. ChatAnthropic/ChatOpenAI use
    @lru_cache'd shared httpx clients that must NOT be closed."""

    def test_skips_non_ollama_module(self):
        llm = Mock(spec=[])
        non_ollama = type("OpenAIClient", (), {})()
        non_ollama._client = Mock()
        type(non_ollama).__module__ = "openai"  # not "ollama"
        llm._client = non_ollama
        llm._async_client = None

        _close_base_llm(llm)

        non_ollama._client.close.assert_not_called()

    def test_delegates_to_wrapper_close_method(self):
        # If the LLM type defines close(), delegate to that and skip
        # introspection. (Wrappers like ProcessingLLMWrapper take this path.)
        class FakeWrapper:
            def __init__(self):
                self.closed = False

            def close(self):
                self.closed = True

        wrapper = FakeWrapper()
        _close_base_llm(wrapper)

        assert wrapper.closed is True


class TestCloseBaseLLMRealHttpxAsync:
    """Empirical validation against a real ``httpx.AsyncClient`` — covers the
    actual #3816 leak shape: a client created inside one ``asyncio.run``
    (loop A) survives loop A's close and must be released by
    ``_close_base_llm`` spinning a fresh loop B. No Ollama server required.
    """

    @pytest.mark.skipif(
        sys.platform == "win32",
        reason="Linux/macOS-specific FD semantics",
    )
    def test_real_async_client_created_in_closed_loop_is_closed(self):
        async def _make():
            return httpx.AsyncClient()

        async_httpx = asyncio.run(_make())  # loop A constructs, then closes
        assert async_httpx.is_closed is False  # client survived loop A

        async_ollama = type("FakeAsyncOllama", (), {})()
        async_ollama._client = async_httpx
        type(async_ollama).__module__ = "ollama"
        llm = Mock(spec=[])
        llm._client = None
        llm._async_client = async_ollama

        _close_base_llm(llm)

        assert async_httpx.is_closed is True
        assert getattr(async_httpx, "_ldr_closed", False) is True

    @pytest.mark.skipif(
        sys.platform == "win32",
        reason="Linux/macOS-specific FD semantics",
    )
    def test_real_async_client_close_is_idempotent(self):
        async def _make():
            return httpx.AsyncClient()

        async_httpx = asyncio.run(_make())
        async_ollama = type("FakeAsyncOllama", (), {})()
        async_ollama._client = async_httpx
        type(async_ollama).__module__ = "ollama"
        llm = Mock(spec=[])
        llm._client = None
        llm._async_client = async_ollama

        _close_base_llm(llm)
        _close_base_llm(llm)  # sentinel short-circuits; must not raise

        assert async_httpx.is_closed is True

    @pytest.mark.skipif(
        sys.platform == "win32",
        reason="Linux/macOS-specific FD semantics",
    )
    def test_no_fd_growth_across_repeated_close_cycles(self):
        # Unit-level analogue of the PR's manual `lsof | grep -c eventpoll`
        # smoke. A real per-cycle leak (~1 FD/iter as observed in #3816)
        # would push the delta well past the +2 slack we leave for ambient
        # logging/pytest churn.
        async def _make():
            return httpx.AsyncClient()

        gc.collect()
        before = _open_fd_count()

        for _ in range(5):
            async_httpx = asyncio.run(_make())
            async_ollama = type("FakeAsyncOllama", (), {})()
            async_ollama._client = async_httpx
            type(async_ollama).__module__ = "ollama"
            llm = Mock(spec=[])
            llm._client = None
            llm._async_client = async_ollama

            _close_base_llm(llm)

            del llm, async_ollama, async_httpx
            gc.collect()

        gc.collect()
        after = _open_fd_count()

        assert after - before <= 2, (
            f"FD count climbed across close cycles: "
            f"before={before}, after={after}"
        )


class TestCloseBaseLLMRealLangchain:
    """End-to-end regression coverage against the real langchain LLM classes
    users actually instantiate. Guards against:

    - A future ``langchain_ollama`` version reshaping ``_client`` /
      ``_async_client`` so the introspection silently no-ops and the
      eventpoll-FD leak from #3816 returns.
    - The introspection accidentally tripping on a non-Ollama LLM and
      closing a shared, lru_cache'd httpx client (which would brick the
      module-global pool for all other instances).

    Construction is no-network: ``ChatOllama(host=…)`` and
    ``ChatOpenAI(api_key=…)`` are both inert until first ``invoke`` /
    ``ainvoke``.
    """

    def test_real_chatollama_through_wrapper_closes_both_clients(self):
        from langchain_ollama import ChatOllama

        from local_deep_research.config.llm_config import (
            wrap_llm_without_think_tags,
        )
        from local_deep_research.utilities.resource_utils import safe_close

        llm = ChatOllama(model="test", base_url="http://localhost:1")
        sync_httpx = llm._client._client
        async_httpx = llm._async_client._client
        assert sync_httpx.is_closed is False
        assert async_httpx.is_closed is False

        wrapper = wrap_llm_without_think_tags(llm, settings_snapshot={})

        safe_close(wrapper, "regression test ChatOllama")

        # Both inner httpx clients must be closed by the chain
        # (safe_close → ProcessingLLMWrapper.close → _close_base_llm →
        # sync close + asyncio.run(aclose)). This is the load-bearing
        # assertion for #3816.
        assert sync_httpx.is_closed is True
        assert async_httpx.is_closed is True

    def test_real_chatopenai_through_wrapper_does_not_close_shared_client(
        self,
    ):
        from langchain_openai import ChatOpenAI

        from local_deep_research.config.llm_config import (
            wrap_llm_without_think_tags,
        )
        from local_deep_research.utilities.resource_utils import safe_close

        llm = ChatOpenAI(model="gpt-4", api_key="sk-test")

        # ChatOpenAI's underlying httpx clients live behind the openai SDK
        # (langchain_openai's lru_cache'd ``_SyncHttpxClientWrapper`` /
        # ``_AsyncHttpxClientWrapper``). They are shared across every
        # ChatOpenAI instance — closing one would brick all the others.
        sync_httpx = llm.root_client._client
        async_httpx = llm.root_async_client._client
        assert sync_httpx.is_closed is False
        assert async_httpx.is_closed is False

        wrapper = wrap_llm_without_think_tags(llm, settings_snapshot={})

        safe_close(wrapper, "regression test ChatOpenAI")

        # ChatOpenAI exposes neither ``_client`` nor ``_async_client``;
        # `_close_base_llm`'s introspection short-circuits cleanly and
        # the shared cached httpx clients must remain open.
        assert sync_httpx.is_closed is False
        assert async_httpx.is_closed is False

    def test_full_wrapper_stack_via_rate_limited_closes_both_clients(self):
        """Regression: when rate limiting is enabled, the production
        wrapper stack is ``ProcessingLLMWrapper(RateLimitedLLMWrapper(
        ChatOllama))`` and ``safe_close`` has to recurse through *both*
        wrapper layers to reach ``_close_base_llm``'s introspection.

        The other ``Real Langchain`` tests only exercise the
        ``ProcessingLLMWrapper`` layer. This one specifically guards
        the ``RateLimitedLLMWrapper.close → _close_base_llm`` hop so a
        future change to that wrapper's ``close()`` doesn't silently
        break the chain and let the #3816 leak come back.
        """
        from langchain_ollama import ChatOllama

        from local_deep_research.config.llm_config import (
            wrap_llm_without_think_tags,
        )
        from local_deep_research.utilities.resource_utils import safe_close
        from local_deep_research.web_search_engines.rate_limiting.llm import (
            create_rate_limited_llm_wrapper,
        )

        llm = ChatOllama(model="test", base_url="http://localhost:1")
        sync_httpx = llm._client._client
        async_httpx = llm._async_client._client
        assert sync_httpx.is_closed is False
        assert async_httpx.is_closed is False

        # Build the stack manually — wrap_llm_without_think_tags only
        # adds the rate-limited layer when ``rate_limiting.llm_enabled``
        # is True in the settings, and we want this test to hold
        # regardless of that setting.
        rate_limited = create_rate_limited_llm_wrapper(llm, provider="ollama")
        full = wrap_llm_without_think_tags(rate_limited, settings_snapshot={})

        safe_close(full, "regression test full wrapper stack")

        # Recursion: safe_close(Processing) → Processing.close →
        # _close_base_llm(RateLimited) → hasattr(type, "close") fires →
        # RateLimited.close() → _close_base_llm(ChatOllama) →
        # introspection closes both inner httpx clients.
        assert sync_httpx.is_closed is True
        assert async_httpx.is_closed is True
