test: remove resources leaks detected by pyleak (#10951)

* test leaking

* fixes on unit tests

* more leaks fixes

* rm deps

* cleanup

* more

* better comment

Author: anakin87

pyproject.toml

@@ -153,7 +153,6 @@ dependencies = [
   "pip",                     # mypy needs pip to install missing stub packages
   "ipython",
   "colorama==0.4.6",         # Pipeline checkpoints test
-  "anyio",                   # needed for asynchronous Path testing
 ]
 
 [tool.hatch.envs.test.scripts]

test/components/agents/test_agent_breakpoints.py

@@ -7,7 +7,6 @@
 from pathlib import Path
 from typing import Any
 
-import anyio
 import pytest
 
 from haystack import component
@@ -709,7 +708,8 @@ async def test_resume_from_chat_generator_async(self, agent, tmp_path, monkeypat
         except BreakpointException:
             pass
 
-        snapshot_files = [path async for path in anyio.Path(debug_path).glob("test_agent_chat_generator_*.json")]
+        # we don't use anyio, because its worker threads outlive the test and leak
+        snapshot_files = list(Path(debug_path).glob("test_agent_chat_generator_*.json"))  # noqa: ASYNC230, ASYNC240
         assert len(snapshot_files) > 0
         latest_snapshot_file = str(max(snapshot_files, key=os.path.getctime))
 
@@ -735,7 +735,8 @@ async def test_resume_from_tool_invoker_async(self, agent, tmp_path, monkeypatch
         except BreakpointException:
             pass
 
-        snapshot_files = [path async for path in anyio.Path(debug_path).glob("test_agent_tool_invoker_*.json")]
+        # we don't use anyio, because its worker threads outlive the test and leak
+        snapshot_files = list(Path(debug_path).glob("test_agent_tool_invoker_*.json"))  # noqa: ASYNC230, ASYNC240
 
         assert len(snapshot_files) > 0
         latest_snapshot_file = str(max(snapshot_files, key=os.path.getctime))
@@ -777,7 +778,8 @@ async def test_resume_from_tool_invoker_and_new_breakpoint_async(self, weather_t
         except BreakpointException:
             pass
 
-        snapshot_files = [path async for path in anyio.Path(debug_path).glob("test_agent_tool_invoker_*.json")]
+        # we don't use anyio, because its worker threads outlive the test and leak
+        snapshot_files = list(Path(debug_path).glob("test_agent_tool_invoker_*.json"))  # noqa: ASYNC230, ASYNC240
         assert len(snapshot_files) > 0
         first_snapshot_file = str(max(snapshot_files, key=os.path.getctime))
 
@@ -790,7 +792,8 @@ async def test_resume_from_tool_invoker_and_new_breakpoint_async(self, weather_t
         except BreakpointException:
             pass
 
-        snapshot_files = [path async for path in anyio.Path(debug_path).glob("test_agent_tool_invoker_*.json")]
+        # we don't use anyio, because its worker threads outlive the test and leak
+        snapshot_files = list(Path(debug_path).glob("test_agent_tool_invoker_*.json"))  # noqa: ASYNC230, ASYNC240
         latest_snapshot_file = str(max(snapshot_files, key=os.path.getctime))
 
         # Resume again
@@ -845,7 +848,8 @@ def custom_callback(snapshot):
         assert exc_info.value.pipeline_snapshot_file_path == "async_callback_id"
 
         # Verify no file was saved to disk
-        all_paths = [path async for path in anyio.Path(debug_path).glob("*.json")]
+        # we don't use anyio, because its worker threads outlive the test and leak
+        all_paths = list(Path(debug_path).glob("*.json"))  # noqa: ASYNC230, ASYNC240
         assert all_paths == []
 
     @pytest.mark.asyncio
@@ -880,7 +884,8 @@ def custom_callback(snapshot):
         assert exc_info.value.pipeline_snapshot_file_path == "async_tool_callback_id"
 
         # Verify no file was saved to disk
-        all_paths = [path async for path in anyio.Path(debug_path).glob("*.json")]
+        # we don't use anyio, because its worker threads outlive the test and leak
+        all_paths = list(Path(debug_path).glob("*.json"))  # noqa: ASYNC230, ASYNC240
         assert all_paths == []
 
     @pytest.mark.asyncio

test/components/builders/test_chat_prompt_builder.py

@@ -17,7 +17,6 @@
 from haystack.core.pipeline.pipeline import Pipeline
 from haystack.dataclasses.chat_message import ChatMessage, FileContent, ImageContent, ReasoningContent
 from haystack.dataclasses.document import Document
-from haystack.document_stores.in_memory import InMemoryDocumentStore
 from haystack.utils.jinja2_chat_extension import END_TAG, START_TAG
 
 
@@ -1038,15 +1037,14 @@ def test_variables_correct_with_list_assignment(self):
         assert res["prompt"][0].text == "x=0, y=1\nHello, my name is John!"
 
     @pytest.mark.integration
-    def test_poisoned_document_does_not_inject_image(self):
-        store = InMemoryDocumentStore()
-        store.write_documents([Document(content="Python is a high-level programming language.")])
+    def test_poisoned_document_does_not_inject_image(self, in_memory_doc_store):
+        in_memory_doc_store.write_documents([Document(content="Python is a high-level programming language.")])
 
         fake_b64 = base64.b64encode(b"ATTACKER_PAYLOAD").decode()
         poison = START_TAG + json.dumps({"image": {"base64_image": fake_b64, "mime_type": "image/png"}}) + END_TAG
-        store.write_documents([Document(content=f"Python tips. {poison}")])
+        in_memory_doc_store.write_documents([Document(content=f"Python tips. {poison}")])
 
-        retriever = InMemoryBM25Retriever(document_store=store)
+        retriever = InMemoryBM25Retriever(document_store=in_memory_doc_store)
         docs = retriever.run(query="Python", top_k=10)["documents"]
 
         template = (

test/components/caching/test_url_cache_checker.py

@@ -67,16 +67,15 @@ def test_from_dict_nonexisting_docstore(self):
         with pytest.raises(ImportError, match=r"Failed to deserialize 'document_store':.*Nonexisting\.DocumentStore"):
             CacheChecker.from_dict(data)
 
-    def test_run(self):
-        docstore = InMemoryDocumentStore()
+    def test_run(self, in_memory_doc_store):
         documents = [
             Document(content="doc1", meta={"url": "https://example.com/1"}),
             Document(content="doc2", meta={"url": "https://example.com/2"}),
             Document(content="doc3", meta={"url": "https://example.com/1"}),
             Document(content="doc4", meta={"url": "https://example.com/2"}),
         ]
-        docstore.write_documents(documents)
-        checker = CacheChecker(docstore, cache_field="url")
+        in_memory_doc_store.write_documents(documents)
+        checker = CacheChecker(in_memory_doc_store, cache_field="url")
         results = checker.run(items=["https://example.com/1", "https://example.com/5"])
         assert results == {"hits": [documents[0], documents[2]], "misses": ["https://example.com/5"]}
 

test/components/classifiers/test_zero_shot_document_classifier.py

@@ -9,7 +9,6 @@
 from haystack import Document, Pipeline
 from haystack.components.classifiers import TransformersZeroShotDocumentClassifier
 from haystack.components.retrievers import InMemoryBM25Retriever
-from haystack.document_stores.in_memory import InMemoryDocumentStore
 from haystack.utils import ComponentDevice, Secret
 
 
@@ -154,10 +153,9 @@ def test_run(self, del_hf_env_vars):
         assert "classification" not in positive_document.to_dict()
         assert "classification" not in negative_document.to_dict()
 
-    def test_serialization_and_deserialization_pipeline(self):
+    def test_serialization_and_deserialization_pipeline(self, in_memory_doc_store):
         pipeline = Pipeline()
-        document_store = InMemoryDocumentStore()
-        retriever = InMemoryBM25Retriever(document_store=document_store)
+        retriever = InMemoryBM25Retriever(document_store=in_memory_doc_store)
         document_classifier = TransformersZeroShotDocumentClassifier(
             model="cross-encoder/nli-deberta-v3-xsmall", labels=["positive", "negative"]
         )

test/components/embedders/test_openai_document_embedder.py

@@ -2,6 +2,7 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
+import contextlib
 import os
 from unittest.mock import Mock, patch
 
@@ -298,15 +299,14 @@ def test_run(self):
     @pytest.mark.integration
     @pytest.mark.asyncio
     async def test_run_async(self):
+        embedder = OpenAIDocumentEmbedder(
+            model="text-embedding-ada-002", meta_fields_to_embed=["topic"], embedding_separator=" | "
+        )
         docs = [
             Document(content="I love cheese", meta={"topic": "Cuisine"}),
             Document(content="A transformer is a deep learning architecture", meta={"topic": "ML"}),
         ]
 
-        model = "text-embedding-ada-002"
-
-        embedder = OpenAIDocumentEmbedder(model=model, meta_fields_to_embed=["topic"], embedding_separator=" | ")
-
         result = await embedder.run_async(documents=docs)
         documents_with_embeddings = result["documents"]
 
@@ -325,3 +325,7 @@ async def test_run_async(self):
         )
 
         assert result["meta"]["usage"] == {"prompt_tokens": 15, "total_tokens": 15}, "Usage information does not match"
+
+        # Close async client; suppress RuntimeError if the event loop is already closed
+        with contextlib.suppress(RuntimeError):
+            await embedder.async_client.close()

test/components/embedders/test_openai_text_embedder.py

@@ -2,6 +2,7 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
+import contextlib
 import os
 
 import pytest
@@ -204,9 +205,7 @@ def test_run(self):
     @pytest.mark.skipif(os.environ.get("OPENAI_API_KEY", "") == "", reason="OPENAI_API_KEY is not set")
     @pytest.mark.integration
     async def test_run_async(self):
-        model = "text-embedding-ada-002"
-
-        embedder = OpenAITextEmbedder(model=model, prefix="prefix ", suffix=" suffix")
+        embedder = OpenAITextEmbedder(model="text-embedding-ada-002", prefix="prefix ", suffix=" suffix")
         result = await embedder.run_async(text="The food was delicious")
 
         assert len(result["embedding"]) == 1536
@@ -217,3 +216,7 @@ async def test_run_async(self):
         )
 
         assert result["meta"]["usage"] == {"prompt_tokens": 6, "total_tokens": 6}, "Usage information does not match"
+
+        # Close async client; suppress RuntimeError if the event loop is already closed
+        with contextlib.suppress(RuntimeError):
+            await embedder.async_client.close()

test/components/extractors/image/test_llm_document_content_extractor.py

@@ -14,7 +14,6 @@
 from haystack.components.writers import DocumentWriter
 from haystack.core.serialization import component_to_dict
 from haystack.dataclasses.chat_message import ChatMessage, ImageContent
-from haystack.document_stores.in_memory import InMemoryDocumentStore
 
 
 class TestLLMDocumentContentExtractor:
@@ -403,18 +402,17 @@ def test_run_on_thread_with_none_prompt(self, monkeypatch):
         not os.environ.get("OPENAI_API_KEY", None),
         reason="Export an env var called OPENAI_API_KEY containing the OpenAI API key to run this test.",
     )
-    def test_live_run(self):
+    def test_live_run(self, in_memory_doc_store):
         docs = [Document(content="", meta={"file_path": "./test/test_files/images/apple.jpg"})]
-        doc_store = InMemoryDocumentStore()
         extractor = LLMDocumentContentExtractor(chat_generator=OpenAIChatGenerator(model="gpt-4.1-nano"))
-        writer = DocumentWriter(document_store=doc_store)
+        writer = DocumentWriter(document_store=in_memory_doc_store)
         pipeline = Pipeline()
         pipeline.add_component("extractor", extractor)
         pipeline.add_component("doc_writer", writer)
         pipeline.connect("extractor.documents", "doc_writer.documents")
         pipeline.run(data={"documents": docs})
 
-        doc_store_docs = doc_store.filter_documents()
+        doc_store_docs = in_memory_doc_store.filter_documents()
         assert len(doc_store_docs) >= 1
         assert len(doc_store_docs[0].content) > 0
 
@@ -423,7 +421,7 @@ def test_live_run(self):
         not os.environ.get("OPENAI_API_KEY", None),
         reason="Export an env var called OPENAI_API_KEY containing the OpenAI API key to run this test.",
     )
-    def test_live_run_on_image_with_metadata(self):
+    def test_live_run_on_image_with_metadata(self, in_memory_doc_store):
         """
         Live test using image_metadata.png: single prompt; LLM can return JSON with document_content
         and metadata keys (author, date, document_type, topic) in one response.
@@ -469,7 +467,6 @@ def test_live_run_on_image_with_metadata(self):
 
         image_path = "./test/test_files/images/image_metadata.png"
         docs = [Document(content="", meta={"file_path": image_path})]
-        doc_store = InMemoryDocumentStore()
         extractor = LLMDocumentContentExtractor(
             prompt=prompt,
             chat_generator=OpenAIChatGenerator(
@@ -494,14 +491,14 @@ def test_live_run_on_image_with_metadata(self):
                 },
             ),
         )
-        writer = DocumentWriter(document_store=doc_store)
+        writer = DocumentWriter(document_store=in_memory_doc_store)
         pipeline = Pipeline()
         pipeline.add_component("extractor", extractor)
         pipeline.add_component("doc_writer", writer)
         pipeline.connect("extractor.documents", "doc_writer.documents")
         pipeline.run(data={"documents": docs})
 
-        doc_store_docs = doc_store.filter_documents()
+        doc_store_docs = in_memory_doc_store.filter_documents()
         assert len(doc_store_docs) >= 1
         doc = doc_store_docs[0]
         assert len(doc.content) > 0, "Expected non-empty content (image/document description)"

test/components/extractors/test_llm_metadata_extractor.py

@@ -12,7 +12,6 @@
 from haystack.components.generators.chat import OpenAIChatGenerator
 from haystack.components.writers import DocumentWriter
 from haystack.dataclasses import ChatMessage
-from haystack.document_stores.in_memory import InMemoryDocumentStore
 
 
 class TestLLMMetadataExtractor:
@@ -264,7 +263,7 @@ def test_run_with_document_content_none(self, monkeypatch):
         not os.environ.get("OPENAI_API_KEY", None),
         reason="Export an env var called OPENAI_API_KEY containing the OpenAI API key to run this test.",
     )
-    def test_live_run(self):
+    def test_live_run(self, in_memory_doc_store):
         docs = [
             Document(content="deepset was founded in 2018 in Berlin, and is known for its Haystack framework"),
             Document(
@@ -310,7 +309,6 @@ def test_live_run(self):
 output:
 """  # noqa: E501
 
-        doc_store = InMemoryDocumentStore()
         extractor = LLMMetadataExtractor(
             prompt=ner_prompt,
             expected_keys=["entities"],
@@ -345,14 +343,14 @@ def test_live_run(self):
                 },
             ),
         )
-        writer = DocumentWriter(document_store=doc_store)
+        writer = DocumentWriter(document_store=in_memory_doc_store)
         pipeline = Pipeline()
         pipeline.add_component("extractor", extractor)
         pipeline.add_component("doc_writer", writer)
         pipeline.connect("extractor.documents", "doc_writer.documents")
         pipeline.run(data={"documents": docs})
 
-        doc_store_docs = doc_store.filter_documents()
+        doc_store_docs = in_memory_doc_store.filter_documents()
         assert len(doc_store_docs) == 2
         assert "entities" in doc_store_docs[0].meta
         assert "entities" in doc_store_docs[1].meta

test/components/generators/chat/test_azure.py

@@ -2,6 +2,7 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
+import contextlib
 import json
 import os
 from typing import Any
@@ -568,14 +569,17 @@ def test_init_should_also_create_async_client_with_same_args(self, tools):
     )
     @pytest.mark.asyncio
     async def test_live_run_async(self):
-        chat_messages = [ChatMessage.from_user("What's the capital of France")]
         component = AzureOpenAIChatGenerator(generation_kwargs={"n": 1})
+        chat_messages = [ChatMessage.from_user("What's the capital of France")]
         results = await component.run_async(chat_messages)
         assert len(results["replies"]) == 1
         message: ChatMessage = results["replies"][0]
         assert "Paris" in message.text
         assert "gpt-4.1-mini" in message.meta["model"]
         assert message.meta["finish_reason"] == "stop"
+        # Close async client; suppress RuntimeError if the event loop is already closed
+        with contextlib.suppress(RuntimeError):
+            await component.async_client.close()
 
     @pytest.mark.integration
     @pytest.mark.skipif(
@@ -588,8 +592,8 @@ async def test_live_run_async(self):
     )
     @pytest.mark.asyncio
     async def test_live_run_with_tools_async(self, tools):
-        chat_messages = [ChatMessage.from_user("What's the weather like in Paris?")]
         component = AzureOpenAIChatGenerator(tools=tools)
+        chat_messages = [ChatMessage.from_user("What's the weather like in Paris?")]
         results = await component.run_async(chat_messages)
         assert len(results["replies"]) == 1
         message = results["replies"][0]
@@ -603,4 +607,8 @@ async def test_live_run_with_tools_async(self, tools):
         assert tool_call.arguments == {"city": "Paris"}
         assert message.meta["finish_reason"] == "tool_calls"
 
+        # Close async client; suppress RuntimeError if the event loop is already closed
+        with contextlib.suppress(RuntimeError):
+            await component.async_client.close()
+
     # additional tests intentionally omitted as they are covered by test_openai.py