Files
wehub-resource-sync 7a0da7932b
OSV-Scanner (Scheduled) / scan-scheduled (push) Failing after 0s
Create Release / test-gate (push) Has been cancelled
Create Release / release-gate (push) Has been cancelled
Create Release / ci-gate (push) Has been cancelled
Create Release / version-check (push) Has been cancelled
Create Release / e2e-test-gate (push) Has been cancelled
Create Release / responsive-test-gate (push) Has been cancelled
Create Release / compat-test-gate (push) Has been cancelled
Create Release / compose-integration-gate (push) Has been cancelled
Create Release / vulture-gate (push) Has been cancelled
Create Release / build (push) Has been cancelled
Create Release / provenance (push) Has been cancelled
Create Release / prerelease-docker (push) Has been cancelled
Create Release / publish-docker (push) Has been cancelled
Create Release / create-release (push) Has been cancelled
Create Release / cleanup-changelog (push) Has been cancelled
Create Release / trigger-pypi (push) Has been cancelled
Create Release / monitor-pypi (push) Has been cancelled
Create Release / Clean up orphan prerelease tags and signatures (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-form] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-metrics] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-workflow] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [settings-core] (push) Has been cancelled
CodeQL Advanced / Analyze (javascript-typescript) (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [history-news] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [library] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [link-analytics] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [chat-core] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [chat-lifecycle] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [error-benchmark] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [settings-pages] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) (push) Has been cancelled
Docker Tests (Consolidated) / Accessibility Tests (push) Has been cancelled
Docker Tests (Consolidated) / LLM Unit Tests (push) Has been cancelled
Docker Tests (Consolidated) / LLM Example Tests (push) Has been cancelled
Docker Tests (Consolidated) / Production Image Smoke Test (push) Has been cancelled
Docker Tests (Consolidated) / Infrastructure Tests (push) Has been cancelled
OSSF Scorecard / OSSF Security Scorecard Analysis (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [mobile] (push) Has been cancelled
Backwards Compatibility / Verify Encryption Constants (push) Has been cancelled
Backwards Compatibility / PyPI Version Compatibility (push) Has been cancelled
Backwards Compatibility / Database Migration Tests (push) Has been cancelled
CodeQL Advanced / Analyze (python) (push) Has been cancelled
Docker Tests (Consolidated) / detect-changes (push) Has been cancelled
Docker Tests (Consolidated) / Build Test Image (push) Has been cancelled
Docker Tests (Consolidated) / All Pytest Tests + Coverage (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [accessibility] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [api-crud] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-login] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-pages] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-register] (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 13:08:55 +08:00

855 lines
31 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""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, and via a brief daemon thread when one is."""
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_closes_async_inside_running_loop_via_thread(self):
"""Regression for the v1.6.10 leak. ``_close_base_llm`` used to
skip the async close when invoked inside a running asyncio loop
and rely on a non-existent "loop owner" cleanup — so the inner
``httpx.AsyncClient`` (and its ``epoll_create`` FD) was silently
abandoned. The current implementation must run the close in a
brief daemon thread whose own ``asyncio.run`` is independent of
the caller's loop.
"""
called = {"count": 0}
async def fake_aclose():
called["count"] += 1
llm = _make_fake_chat_ollama(async_aclose=fake_aclose)
async def driver():
_close_base_llm(llm)
asyncio.run(driver())
# aclose ran exactly once (via the cleanup thread, not skipped).
assert called["count"] == 1
# And _ldr_closed IS set on success — subsequent calls short-circuit.
async_httpx = llm._async_client._client
assert async_httpx._ldr_closed is True
def test_in_loop_close_is_idempotent(self):
"""A close fired from inside a loop should set ``_ldr_closed`` just
like the no-loop path, so repeat calls don't re-spawn the cleanup
thread or re-run ``aclose``."""
called = {"count": 0}
async def fake_aclose():
called["count"] += 1
llm = _make_fake_chat_ollama(async_aclose=fake_aclose)
async def driver():
_close_base_llm(llm)
_close_base_llm(llm)
_close_base_llm(llm)
asyncio.run(driver())
assert called["count"] == 1
def test_in_loop_close_timeout_does_not_mark_closed(self):
"""If the cleanup thread is still alive after the 5-second join
(e.g. ``aclose`` is blocked on a stuck server), the sentinel must
NOT be set so a later call can retry — and the FD leak is at
least visible via WARNING log instead of silent drift.
"""
import threading
from unittest.mock import patch
release = threading.Event()
aclose_started = threading.Event()
async def slow_aclose():
aclose_started.set()
# Block until released (or the test's shortened join fires).
await asyncio.get_event_loop().run_in_executor(
None, release.wait, 30
)
llm = _make_fake_chat_ollama(async_aclose=slow_aclose)
original_thread = threading.Thread
class _ShortJoinThread(original_thread):
def join(self, timeout=None):
# Tighten the production 5s wait to 200ms for the test
# so we don't actually sit here for 5 seconds.
return super().join(timeout=0.2)
try:
with patch("threading.Thread", _ShortJoinThread):
async def driver():
_close_base_llm(llm)
asyncio.run(driver())
finally:
release.set()
# ``aclose`` started but the join timed out before it could
# finish — sentinel must be unset so the FD is not silently
# leaked (a subsequent _close_base_llm call should retry).
assert aclose_started.is_set()
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.fd_canary
@pytest.mark.timeout(180)
@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 +8 slack.
#
# Sizing rationale (#4214 + #3816 follow-up):
# Each cycle invokes ``asyncio.run`` twice — once to construct the
# client in a (then-closed) loop A, once inside ``_close_base_llm``
# to spin loop B for ``aclose()``. On Py 3.14 these transient loops
# leave ambient FDs from selector/self-pipe lazy release. CI on the
# 5-iter version of this test showed a deterministic +3 delta (58
# of 58 recent failures), i.e. ~0.6 FDs/cycle steady-state — not
# absorbable by a single warmup since it's per-cycle, not init-only.
# At 20 iterations the expected ceiling is ~0.6 × 20 = ~12 in the
# worst case (observed ~34 locally); the +8 slack covers the
# typical range without masking a real per-iteration leak, which
# would land at ≥+20 (1 FD/iter, #3816 shape) + ambient → far
# above the threshold.
async def _make():
return httpx.AsyncClient()
def _one_cycle():
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
# Warmup: one cycle before measuring absorbs one-time init drift
# (lazy imports, logging handlers, asyncio internals first-init).
_one_cycle()
gc.collect()
before = _open_fd_count()
for _ in range(20):
_one_cycle()
gc.collect()
gc.collect()
after = _open_fd_count()
assert after - before <= 8, (
f"FD count climbed across close cycles: "
f"before={before}, after={after}"
)
@pytest.mark.fd_canary
@pytest.mark.timeout(120)
@pytest.mark.skipif(
sys.platform == "win32",
reason="Linux/macOS-specific FD semantics",
)
def test_no_fd_growth_when_closed_inside_running_loop(self):
"""The user-facing regression. Before the fix, calling
``_close_base_llm`` inside a running loop silently skipped the
async close — every iteration leaked ~1 ``epoll_create`` FD.
After the fix, the cleanup thread closes the client even when
a loop is running, so the FD count stays flat across iterations.
"""
async def _close_from_inside_loop():
async_httpx = httpx.AsyncClient()
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 is invoked while this loop is running —
# the exact scenario that leaked before the fix.
_close_base_llm(llm)
assert async_httpx.is_closed is True
# Warmup cycle to absorb one-time init drift (cleanup-thread
# startup, lazy imports) — see the sibling repeated-close test.
asyncio.run(_close_from_inside_loop())
gc.collect()
before = _open_fd_count()
for _ in range(5):
asyncio.run(_close_from_inside_loop())
gc.collect()
gc.collect()
after = _open_fd_count()
assert after - before <= 2, (
f"FD count climbed across in-loop close cycles "
f"(this is the #3816-shaped leak): "
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
class TestCloseBaseLLMRealOllamaEmbeddings:
"""End-to-end regression coverage for ``OllamaEmbeddings``.
After the langchain_community → langchain_ollama migration
(#4352/#4353), ``OllamaEmbeddings`` carries the same
``_client`` / ``_async_client`` shape as ``ChatOllama`` — eagerly
constructed by a Pydantic ``@model_validator(mode="after")``, with no
``close()`` / ``aclose()`` / ``__del__`` of its own. The deprecated
``langchain_community.embeddings.OllamaEmbeddings`` was FD-safe by
accident (``requests.post()`` per call, no persistent client); the
new class is not. The resource-cleanup doc predicted exactly this
leak shape; this class is the canary that catches a recurrence if
a future migration breaks the close path again.
Construction is no-network: ``OllamaEmbeddings(base_url=…)`` is inert
until the first ``embed_query`` / ``embed_documents`` call.
"""
def test_real_ollama_embeddings_closes_both_clients(self):
from langchain_ollama import OllamaEmbeddings
embeddings = OllamaEmbeddings(
model="test", base_url="http://localhost:1"
)
sync_httpx = embeddings._client._client
async_httpx = embeddings._async_client._client
assert sync_httpx.is_closed is False
assert async_httpx.is_closed is False
_close_base_llm(embeddings)
# Both inner httpx clients must be closed. Sync side is closed
# synchronously; async side via the ``asyncio.run`` branch (no
# running loop in this thread).
assert sync_httpx.is_closed is True
assert async_httpx.is_closed is True
def test_real_ollama_embeddings_close_is_idempotent(self):
from langchain_ollama import OllamaEmbeddings
embeddings = OllamaEmbeddings(
model="test", base_url="http://localhost:1"
)
async_httpx = embeddings._async_client._client
_close_base_llm(embeddings)
_close_base_llm(embeddings) # sentinel short-circuits; must not raise
assert async_httpx.is_closed is True
assert getattr(async_httpx, "_ldr_closed", False) is True
def test_local_embedding_manager_close_closes_ollama_embeddings(self):
"""Integration: ``LocalEmbeddingManager.close()`` must flow through
to ``_close_base_llm`` for an Ollama-backed embeddings instance.
Without this, the ``LibraryRAGService`` close path — which
cascades to the manager's ``close()`` — leaks the underlying
httpx clients per RAG request (the #3816-shaped FD ramp, now on
the embeddings side).
"""
from langchain_ollama import OllamaEmbeddings
from local_deep_research.web_search_engines.engines.local_embedding_manager import (
LocalEmbeddingManager,
)
mgr = LocalEmbeddingManager(
embedding_model="test",
embedding_model_type="ollama",
ollama_base_url="http://localhost:1",
settings_snapshot={},
)
# Inject a real embeddings instance directly to bypass the lazy
# ``embeddings`` property — its provider lookup would touch the
# settings / database layers, which aren't relevant to this test.
mgr._embeddings = OllamaEmbeddings(
model="test", base_url="http://localhost:1"
)
sync_httpx = mgr._embeddings._client._client
async_httpx = mgr._embeddings._async_client._client
assert sync_httpx.is_closed is False
assert async_httpx.is_closed is False
mgr.close()
assert sync_httpx.is_closed is True
assert async_httpx.is_closed is True
def test_local_embedding_manager_close_is_safe_for_non_ollama(self):
"""The close path must not blow up for non-Ollama providers. The
module-prefix check inside ``_close_base_llm`` ensures the
``HuggingFaceEmbeddings`` fallback (and any other non-Ollama
provider) is a no-op rather than an AttributeError.
"""
from local_deep_research.web_search_engines.engines.local_embedding_manager import (
LocalEmbeddingManager,
)
mgr = LocalEmbeddingManager(
embedding_model="test",
embedding_model_type="sentence_transformers",
settings_snapshot={},
)
# Stand-in with neither ``_client`` nor ``_async_client``.
non_ollama = type("FakeHFEmbeddings", (), {})()
mgr._embeddings = non_ollama
# Must complete cleanly; the introspection short-circuits when
# the attributes are absent. ``hasattr(type(non_ollama), "close")``
# is False for our minimal type, so the wrapper-delegation path
# is also skipped — exactly the behaviour we want.
mgr.close()
assert mgr._embeddings is None
assert mgr._closed is True
class TestLibraryRAGServiceCloseOwnership:
"""``LibraryRAGService`` may be constructed with a caller-supplied
``embedding_manager`` (test fixtures, multi-service callers reusing
one manager). Closing the service must NOT close a manager it didn't
create — that would burn the caller's reference and surface as a
use-after-close error in the next call site.
"""
def test_close_does_not_touch_caller_supplied_manager(self):
from unittest.mock import Mock
from local_deep_research.research_library.services.library_rag_service import (
LibraryRAGService,
)
external_manager = Mock(spec=["close"])
# Bypass the heavy ``__init__`` (DB session, FileIntegrityManager,
# text splitter) — only the ownership + close branch matters here.
svc = LibraryRAGService.__new__(LibraryRAGService)
svc._closed = False
svc.embedding_manager = external_manager
svc._owns_embedding_manager = False
svc.faiss_index = None
svc.rag_index_record = None
svc.integrity_manager = None
svc.text_splitter = None
svc.close()
external_manager.close.assert_not_called()
assert svc.embedding_manager is None
def test_close_tears_down_owned_manager(self):
from unittest.mock import Mock
from local_deep_research.research_library.services.library_rag_service import (
LibraryRAGService,
)
owned_manager = Mock(spec=["close"])
svc = LibraryRAGService.__new__(LibraryRAGService)
svc._closed = False
svc.embedding_manager = owned_manager
svc._owns_embedding_manager = True
svc.faiss_index = None
svc.rag_index_record = None
svc.integrity_manager = None
svc.text_splitter = None
svc.close()
owned_manager.close.assert_called_once()
assert svc.embedding_manager is None
class TestInLoopCleanupThreadContract:
"""The in-running-loop branch of ``_close_base_llm`` hands the
``aclose()`` off to a daemon thread (see the
``else: # A loop is running in this thread`` block that spawns
a ``ldr-async-llm-close`` thread). Two properties of that thread
are load-bearing but not pinned by the other tests in this file:
1. ``daemon=True`` — without this, a cleanup thread blocked on a
stuck server would prevent Python interpreter shutdown.
2. The thread is given a recognizable name so operators chasing FD
leaks can correlate ``threading.enumerate()`` / ``top`` output
against the leak source.
Plus the third gap: when the cleanup thread completes BUT its inner
``asyncio.run(aclose())`` raised, the main thread still sets
``_ldr_closed = True`` in the ``else`` branch reached when
``t.is_alive()`` is False. The pre-existing
``test_swallows_async_close_exception`` covers this for the no-loop
branch; this class covers the same invariant for the in-loop branch.
"""
def test_cleanup_thread_is_daemon_so_shutdown_is_not_blocked(self):
"""Capture the thread object created by the running-loop branch
and assert ``daemon`` is True. A non-daemon cleanup thread
blocked on a slow server would hang interpreter shutdown until
the OS kills the process — exactly the failure mode the
docstring of ``_close_base_llm`` warns about when motivating
the brief daemon thread.
"""
import threading
from unittest.mock import patch
async def fake_aclose():
return None
llm = _make_fake_chat_ollama(async_aclose=fake_aclose)
captured = {}
original_thread = threading.Thread
class _CapturingThread(original_thread):
def __init__(self, *args, **kwargs):
captured["daemon"] = kwargs.get("daemon")
captured["name"] = kwargs.get("name")
super().__init__(*args, **kwargs)
async def driver():
with patch("threading.Thread", _CapturingThread):
_close_base_llm(llm)
asyncio.run(driver())
assert captured["daemon"] is True, (
"cleanup thread must be a daemon — otherwise a stuck "
"aclose() would block Python interpreter shutdown"
)
# Name is operator-facing for FD-leak triage; pin a stable
# prefix rather than the exact string so future tweaks can
# extend without breaking the test.
assert captured["name"] is not None
assert "ldr" in captured["name"].lower()
def test_in_loop_close_marks_closed_even_when_inner_aclose_raises(
self,
):
"""The cleanup thread's ``_close_in_thread`` runs
``asyncio.run(aclose())`` inside a ``try/except Exception`` — if
``aclose`` raises, the thread logs a warning and exits cleanly.
The main thread sees ``t.is_alive() == False`` and falls into
the ``else`` branch that sets ``_ldr_closed = True``. Pin that
invariant: the sentinel is the signal to never retry a
known-broken close, regardless of the branch (no-loop or
in-loop) that performed it.
"""
async def boom_aclose():
raise RuntimeError("simulated upstream failure")
llm = _make_fake_chat_ollama(async_aclose=boom_aclose)
async def driver():
_close_base_llm(llm)
asyncio.run(driver())
async_httpx = llm._async_client._client
assert async_httpx._ldr_closed is True, (
"Even when aclose raises inside the cleanup thread, the "
"main thread must mark the client closed to prevent "
"indefinite retries on a known-broken close"
)