项目文件夹

文件
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 行
31 KiB
Python

此文件含有模棱两可的 Unicode 字符
此文件含有可能会与其他字符混淆的 Unicode 字符。 如果您是想特意这样的,可以安全地忽略该警告。 使用 Escape 按钮显示他们。
"""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 ~3–4 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"
)