langchain: Use Blockbuster to detect blocking calls in asyncio during tests (#29616)

Same as #29043 for the
langchain package.

**Dependencies:**
- blockbuster (test)

**Twitter handle:** cbornet_

---------

Co-authored-by: Erick Friis <[email protected]>

Author: cbornet

libs/langchain/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "pdm.backend"
 
 [project]
 authors = []
-license = {text = "MIT"}
+license = { text = "MIT" }
 requires-python = "<4.0,>=3.9"
 dependencies = [
     "langchain-core<1.0.0,>=0.3.34",
@@ -66,6 +66,7 @@ test = [
     "syrupy<5.0.0,>=4.0.2",
     "requests-mock<2.0.0,>=1.11.0",
     "pytest-xdist<4.0.0,>=3.6.1",
+    "blockbuster<1.6,>=1.5.14",
     "cffi<1.17.1; python_version < \"3.10\"",
     "cffi; python_version >= \"3.10\"",
     "langchain-tests @ file:///${PROJECT_ROOT}/../standard-tests",
@@ -75,9 +76,7 @@ test = [
     "toml>=0.10.2",
     "packaging>=24.2",
 ]
-codespell = [
-    "codespell<3.0.0,>=2.2.0",
-]
+codespell = ["codespell<3.0.0,>=2.2.0"]
 test_integration = [
     "pytest-vcr<2.0.0,>=1.0.2",
     "urllib3<2; python_version < \"3.10\"",
@@ -116,31 +115,39 @@ dev = [
 
 [tool.ruff]
 target-version = "py39"
-exclude = [ "tests/integration_tests/examples/non-utf8-encoding.py",]
+exclude = ["tests/integration_tests/examples/non-utf8-encoding.py"]
 
 [tool.mypy]
 ignore_missing_imports = "True"
 disallow_untyped_defs = "True"
-exclude = [ "notebooks", "examples", "example_data",]
+exclude = ["notebooks", "examples", "example_data"]
 
 [tool.codespell]
 skip = ".git,*.pdf,*.svg,*.pdf,*.yaml,*.ipynb,poetry.lock,*.min.js,*.css,package-lock.json,example_data,_dist,examples,*.trig"
 ignore-regex = ".*(Stati Uniti|Tense=Pres).*"
 ignore-words-list = "momento,collison,ned,foor,reworkd,parth,whats,aapply,mysogyny,unsecure,damon,crate,aadd,symbl,precesses,accademia,nin"
 
 [tool.ruff.lint]
-select = [ "E", "F", "I", "T201", "D",]
+select = ["E", "F", "I", "T201", "D"]
 pydocstyle = { convention = "google" }
 
 [tool.ruff.lint.per-file-ignores]
 "tests/*" = ["D"]
 "!langchain/indexes/vectorstore.py" = ["D"]
 
 [tool.coverage.run]
-omit = [ "tests/*",]
+omit = ["tests/*"]
 
 [tool.pytest.ini_options]
 addopts = "--strict-markers --strict-config --durations=5 --snapshot-warn-unused -vv"
-markers = [ "requires: mark tests as requiring a specific library", "scheduled: mark tests to run in scheduled testing", "compile: mark placeholder test used to compile integration tests without running them",]
+markers = [
+    "requires: mark tests as requiring a specific library",
+    "scheduled: mark tests to run in scheduled testing",
+    "compile: mark placeholder test used to compile integration tests without running them",
+]
 asyncio_mode = "auto"
-filterwarnings = [ "ignore::langchain_core._api.beta_decorator.LangChainBetaWarning", "ignore::langchain_core._api.deprecation.LangChainDeprecationWarning:tests", "ignore::langchain_core._api.deprecation.LangChainPendingDeprecationWarning:tests",]
+filterwarnings = [
+    "ignore::langchain_core._api.beta_decorator.LangChainBetaWarning",
+    "ignore::langchain_core._api.deprecation.LangChainDeprecationWarning:tests",
+    "ignore::langchain_core._api.deprecation.LangChainPendingDeprecationWarning:tests",
+]

libs/langchain/tests/unit_tests/agents/test_agent.py

@@ -1,5 +1,6 @@
 """Unit tests for agents."""
 
+import asyncio
 import json
 from itertools import cycle
 from typing import Any, Dict, List, Optional, Union, cast
@@ -465,33 +466,31 @@ def fake_parse(inputs: dict) -> Union[AgentFinish, AgentAction]:
     executor = AgentExecutor(agent=agent, tools=[])  # type: ignore[arg-type]
 
     # Invoke
-    result = executor.invoke({"question": "hello"})
+    result: Any = await asyncio.to_thread(executor.invoke, {"question": "hello"})
     assert result == {"foo": "meow", "question": "hello"}
 
     # ainvoke
     result = await executor.ainvoke({"question": "hello"})
     assert result == {"foo": "meow", "question": "hello"}
 
     # Batch
-    result = executor.batch(  # type: ignore[assignment]
-        [{"question": "hello"}, {"question": "hello"}]
+    result = await asyncio.to_thread(
+        executor.batch, [{"question": "hello"}, {"question": "hello"}]
     )
     assert result == [
         {"foo": "meow", "question": "hello"},
         {"foo": "meow", "question": "hello"},
     ]
 
     # abatch
-    result = await executor.abatch(  # type: ignore[assignment]
-        [{"question": "hello"}, {"question": "hello"}]
-    )
+    result = await executor.abatch([{"question": "hello"}, {"question": "hello"}])
     assert result == [
         {"foo": "meow", "question": "hello"},
         {"foo": "meow", "question": "hello"},
     ]
 
     # Stream
-    results = list(executor.stream({"question": "hello"}))
+    results = await asyncio.to_thread(list, executor.stream({"question": "hello"}))
     assert results == [
         {"foo": "meow", "messages": [AIMessage(content="hard-coded-message")]}
     ]
@@ -587,7 +586,7 @@ def find_pet(pet: str) -> str:
     executor = AgentExecutor(agent=agent, tools=[find_pet])  # type: ignore[arg-type, list-item]
 
     # Invoke
-    result = executor.invoke({"question": "hello"})
+    result = await asyncio.to_thread(executor.invoke, {"question": "hello"})
     assert result == {"foo": "meow", "question": "hello"}
 
     # ainvoke
@@ -705,7 +704,7 @@ def pet_pet(pet: str) -> str:
     executor = AgentExecutor(agent=agent, tools=[find_pet])  # type: ignore[arg-type, list-item]
 
     # Invoke
-    result = executor.invoke({"question": "hello"})
+    result = await asyncio.to_thread(executor.invoke, {"question": "hello"})
     assert result == {"foo": "meow", "question": "hello"}
 
     # ainvoke
@@ -859,7 +858,7 @@ def find_pet(pet: str) -> str:
     executor = AgentExecutor(agent=agent, tools=[find_pet])  # type: ignore[arg-type, list-item]
 
     # Invoke
-    result = executor.invoke({"question": "hello"})
+    result = await asyncio.to_thread(executor.invoke, {"question": "hello"})
     assert result == {
         "output": "The cat is spying from under the bed.",
         "question": "hello",
@@ -1066,7 +1065,7 @@ def check_time() -> str:
         executor = AgentExecutor(agent=agent, tools=[find_pet])  # type: ignore[arg-type, list-item]
 
         # Invoke
-        result = executor.invoke({"question": "hello"})
+        result = await asyncio.to_thread(executor.invoke, {"question": "hello"})
         assert result == {
             "output": "The cat is spying from under the bed.",
             "question": "hello",

libs/langchain/tests/unit_tests/conftest.py

@@ -1,12 +1,42 @@
 """Configuration for unit tests."""
 
+from collections.abc import Iterator
 from importlib import util
 from typing import Dict, Sequence
 
 import pytest
+from blockbuster import blockbuster_ctx
 from pytest import Config, Function, Parser
 
 
+@pytest.fixture(autouse=True)
+def blockbuster() -> Iterator[None]:
+    with blockbuster_ctx("langchain") as bb:
+        bb.functions["io.TextIOWrapper.read"].can_block_in(
+            "langchain/__init__.py", "<module>"
+        )
+
+        for func in ["os.stat", "os.path.abspath"]:
+            (
+                bb.functions[func]
+                .can_block_in("langchain_core/runnables/base.py", "__repr__")
+                .can_block_in(
+                    "langchain_core/beta/runnables/context.py", "aconfig_with_context"
+                )
+            )
+
+        for func in ["os.stat", "io.TextIOWrapper.read"]:
+            bb.functions[func].can_block_in(
+                "langsmith/client.py", "_default_retry_config"
+            )
+
+        for bb_function in bb.functions.values():
+            bb_function.can_block_in(
+                "freezegun/api.py", "_get_cached_module_attributes"
+            )
+        yield
+
+
 def pytest_addoption(parser: Parser) -> None:
     """Add custom command line options to pytest."""
     parser.addoption(

libs/langchain/tests/unit_tests/llms/test_fake_chat_model.py

@@ -184,7 +184,12 @@ async def on_llm_new_token(
     model = GenericFakeChatModel(messages=infinite_cycle)
     tokens: List[str] = []
     # New model
-    results = list(model.stream("meow", {"callbacks": [MyCustomAsyncHandler(tokens)]}))
+    results = [
+        chunk
+        async for chunk in model.astream(
+            "meow", {"callbacks": [MyCustomAsyncHandler(tokens)]}
+        )
+    ]
     assert results == [
         _AnyIdAIMessageChunk(content="hello"),
         _AnyIdAIMessageChunk(content=" "),

libs/langchain/tests/unit_tests/test_dependencies.py

@@ -77,6 +77,7 @@ def test_test_group_dependencies(uv_conf: Mapping[str, Any]) -> None:
             "pytest-socket",
             "pytest-watcher",
             "pytest-xdist",
+            "blockbuster",
             "responses",
             "syrupy",
             "toml",