import asyncio import logging from collections.abc import Mapping from pathlib import Path from types import SimpleNamespace import pytest import deerflow.community.ragflow.tools as ragflow_tools from deerflow.community.ragflow.client import RAGFlowAPIError, RAGFlowConnectionError from deerflow.community.ragflow.formatting import format_retrieval_result from deerflow.config.tool_config import ToolConfig from deerflow.tools.tools import get_available_tools DATASET_ID_1 = "0123456789abcdef0123456789abcdef" DATASET_ID_2 = "fedcba9876543210fedcba9876543210" MISSING_DATASET_ID = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" EMBEDDING_V2 = "text-embedding-v2@primary@Tongyi-Qianwen" EMBEDDING_V3 = "text-embedding-v3@primary@Tongyi-Qianwen" def _dataset( dataset_id: str, name: str, *, embedding_model: str = EMBEDDING_V3, chunk_count: int = 1, ) -> dict: return { "id": dataset_id, "name": name, "embedding_model": embedding_model, "chunk_count": chunk_count, } class FakeRAGFlowClient: def __init__( self, *, datasets_by_id: Mapping[str, list[dict]] | None = None, dataset_errors_by_id: Mapping[str, Exception] | None = None, all_datasets: list[dict] | None = None, retrieval: dict | None = None, retrieval_by_dataset_ids: Mapping[tuple[str, ...], dict] | None = None, retrieval_errors_by_dataset_ids: Mapping[tuple[str, ...], Exception] | None = None, error: Exception | None = None, documents_by_dataset_id: Mapping[str, list[dict]] | None = None, ) -> None: self.datasets_by_id = dict(datasets_by_id or {}) self.dataset_errors_by_id = dict(dataset_errors_by_id or {}) self.all_datasets = list(all_datasets or []) self.retrieval = retrieval or {"chunks": [], "doc_aggs": [], "total": 0} self.retrieval_by_dataset_ids = dict(retrieval_by_dataset_ids or {}) self.retrieval_errors_by_dataset_ids = dict(retrieval_errors_by_dataset_ids or {}) self.error = error self.documents_by_dataset_id = dict(documents_by_dataset_id or {}) self.list_calls: list[str | None] = [] self.document_list_calls: list[tuple[str, list[tuple[str, str]]]] = [] self.retrieve_calls: list[tuple[str, dict]] = [] async def list_datasets(self, *, dataset_id: str | None = None) -> list[dict]: if self.error is not None: raise self.error self.list_calls.append(dataset_id) if dataset_id is None: return self.all_datasets if error := self.dataset_errors_by_id.get(dataset_id): raise error return self.datasets_by_id.get(dataset_id, []) async def list_documents( self, dataset_id: str, *, params: list[tuple[str, str]], ) -> dict: self.document_list_calls.append((dataset_id, params)) requested_ids = [value for key, value in params if key == "ids"] documents = self.documents_by_dataset_id.get(dataset_id, []) if requested_ids: documents = [document for document in documents if document.get("id") in requested_ids] return { "code": 0, "data": {"docs": documents, "total": len(documents)}, } async def retrieve(self, query: str, **kwargs: object) -> dict: if self.error is not None: raise self.error self.retrieve_calls.append((query, kwargs)) dataset_ids = kwargs.get("dataset_ids") key = tuple(dataset_ids) if isinstance(dataset_ids, list) else () if error := self.retrieval_errors_by_dataset_ids.get(key): raise error if key in self.retrieval_by_dataset_ids: return self.retrieval_by_dataset_ids[key] return self.retrieval @pytest.fixture(autouse=True) def reset_warning_deduplication() -> None: ragflow_tools._warned.clear() def _config( *, configured: bool = True, api_key: str | None = "ragflow-secret", base_url: str = "http://ragflow.test", datasets: list[str] | None = None, page_size: int = 8, ) -> SimpleNamespace: extra: dict[str, object] = { "base_url": base_url, "api_key": api_key, "timeout": 30, "page_size": page_size, "similarity_threshold": 0.2, "vector_similarity_weight": 0.3, "top_k": 256, "max_chars_per_chunk": 800, "max_total_chars": 8000, } if datasets is not None: extra["datasets"] = datasets search_config = ToolConfig( name="knowledge_search", group="knowledge", use="deerflow.community.ragflow.tools:knowledge_search_tool", **extra, ) return SimpleNamespace( get_tool_config=lambda name: search_config if configured and name == "knowledge_search" else None, ) def _install(monkeypatch: pytest.MonkeyPatch, fake: FakeRAGFlowClient, *, config: SimpleNamespace | None = None) -> None: monkeypatch.setattr(ragflow_tools, "get_app_config", lambda: config or _config(datasets=[DATASET_ID_1])) monkeypatch.setattr(ragflow_tools, "_build_client", lambda settings: fake) @pytest.mark.anyio async def test_knowledge_search_resolves_configured_ids_to_current_names(monkeypatch: pytest.MonkeyPatch) -> None: fake = FakeRAGFlowClient( datasets_by_id={ DATASET_ID_1: [_dataset(DATASET_ID_1, "HR Policies")], DATASET_ID_2: [_dataset(DATASET_ID_2, "Engineering")], }, retrieval={ "chunks": [ { "dataset_id": DATASET_ID_1, "document_id": "doc-1", "document_keyword": "handbook.pdf", "content": "Annual leave is based on years of service.", "similarity": 0.874, } ], "doc_aggs": [{"doc_id": "doc-1", "doc_name": "handbook.pdf", "count": 1}], "total": 1, }, ) _install(monkeypatch, fake, config=_config(datasets=[DATASET_ID_1, DATASET_ID_2])) result = await ragflow_tools.knowledge_search("annual leave") assert fake.list_calls == [DATASET_ID_1, DATASET_ID_2] assert fake.retrieve_calls == [ ( "annual leave", { "dataset_ids": [DATASET_ID_1, DATASET_ID_2], "page_size": 8, "similarity_threshold": 0.2, "vector_similarity_weight": 0.3, "top_k": 256, }, ) ] assert "[1] HR Policies / handbook.pdf (score 0.87)" in result assert "Annual leave" in result assert "Matched documents: handbook.pdf (1 chunk)" in result assert DATASET_ID_1 not in result assert DATASET_ID_2 not in result @pytest.mark.anyio async def test_knowledge_search_uses_id_filter_and_survives_dataset_rename(monkeypatch: pytest.MonkeyPatch) -> None: fake = FakeRAGFlowClient( datasets_by_id={DATASET_ID_1: [_dataset(DATASET_ID_1, "Renamed Policies")]}, retrieval={"chunks": [{"dataset_id": DATASET_ID_1, "document_keyword": "policy.pdf", "content": "Current policy."}]}, ) _install(monkeypatch, fake) result = await ragflow_tools.knowledge_search("leave") assert fake.list_calls == [DATASET_ID_1] assert fake.retrieve_calls[0][1]["dataset_ids"] == [DATASET_ID_1] assert "Renamed Policies / policy.pdf" in result assert DATASET_ID_1 not in result @pytest.mark.anyio async def test_missing_bound_dataset_returns_indexed_operator_guidance( monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, ) -> None: fake = FakeRAGFlowClient( datasets_by_id={DATASET_ID_1: [_dataset(DATASET_ID_1, "Existing")]}, ) _install(monkeypatch, fake, config=_config(datasets=[DATASET_ID_1, MISSING_DATASET_ID])) with caplog.at_level(logging.WARNING, logger="deerflow.community.ragflow.tools"): result = await ragflow_tools.knowledge_search("leave") assert result == "Error: The 2nd entry of knowledge_search.datasets was not found or is inaccessible; check config.yaml." assert MISSING_DATASET_ID not in result assert fake.list_calls == [DATASET_ID_1, MISSING_DATASET_ID] assert fake.retrieve_calls == [] assert MISSING_DATASET_ID in caplog.text assert "code=None" in caplog.text @pytest.mark.anyio async def test_missing_bound_dataset_error_does_not_expose_configured_id(monkeypatch: pytest.MonkeyPatch) -> None: fake = FakeRAGFlowClient() _install(monkeypatch, fake, config=_config(datasets=[MISSING_DATASET_ID])) result = await ragflow_tools.knowledge_search("leave") assert MISSING_DATASET_ID not in result assert "[DATASET_ID]" not in result @pytest.mark.anyio async def test_bound_dataset_api_error_uses_normal_redacted_error_handler( monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, ) -> None: fake = FakeRAGFlowClient( dataset_errors_by_id={ DATASET_ID_1: RAGFlowAPIError("invalid credential ragflow-secret", code=102), } ) _install(monkeypatch, fake, config=_config(datasets=[DATASET_ID_1])) with caplog.at_level(logging.WARNING, logger="deerflow.community.ragflow.tools"): result = await ragflow_tools.knowledge_search("leave") assert result == "Error: invalid credential [REDACTED]" assert "code=102" in caplog.text assert fake.retrieve_calls == [] @pytest.mark.anyio async def test_mismatched_id_filtered_response_returns_operator_guidance(monkeypatch: pytest.MonkeyPatch) -> None: fake = FakeRAGFlowClient(datasets_by_id={DATASET_ID_1: [_dataset(DATASET_ID_2, "Wrong dataset")]}) _install(monkeypatch, fake, config=_config(datasets=[DATASET_ID_1])) result = await ragflow_tools.knowledge_search("leave") assert result == "Error: The 1st entry of knowledge_search.datasets was not found or is inaccessible; check config.yaml." assert fake.retrieve_calls == [] @pytest.mark.anyio async def test_missing_dataset_binding_lists_all_and_passes_every_id(monkeypatch: pytest.MonkeyPatch) -> None: fake = FakeRAGFlowClient( all_datasets=[ _dataset(DATASET_ID_1, "HR Policies"), _dataset(DATASET_ID_2, "Engineering"), ], retrieval={"chunks": [{"dataset_id": DATASET_ID_2, "document_keyword": "guide.pdf", "content": "Build guide."}]}, ) _install(monkeypatch, fake, config=_config(datasets=None)) result = await ragflow_tools.knowledge_search("leave") assert fake.list_calls == [None] assert fake.retrieve_calls[0][1]["dataset_ids"] == [DATASET_ID_1, DATASET_ID_2] assert "Engineering / guide.pdf" in result assert DATASET_ID_1 not in result assert DATASET_ID_2 not in result @pytest.mark.anyio async def test_mixed_embedding_models_are_retrieved_in_parallel_groups_and_rank_interleaved(monkeypatch: pytest.MonkeyPatch) -> None: fake = FakeRAGFlowClient( datasets_by_id={ DATASET_ID_1: [_dataset(DATASET_ID_1, "Legacy", embedding_model=EMBEDDING_V2)], DATASET_ID_2: [_dataset(DATASET_ID_2, "Current", embedding_model=EMBEDDING_V3)], }, retrieval_by_dataset_ids={ (DATASET_ID_1,): { "chunks": [ {"dataset_id": DATASET_ID_1, "document_id": "legacy-1", "document_keyword": "legacy-1.txt", "content": "Legacy rank one.", "similarity": 0.41}, {"dataset_id": DATASET_ID_1, "document_id": "legacy-2", "document_keyword": "legacy-2.txt", "content": "Legacy rank two.", "similarity": 0.99}, ], "doc_aggs": [ {"doc_id": "legacy-1", "doc_name": "legacy-1.txt", "count": 1}, {"doc_id": "legacy-2", "doc_name": "legacy-2.txt", "count": 1}, ], "total": 2, }, (DATASET_ID_2,): { "chunks": [ {"dataset_id": DATASET_ID_2, "document_id": "current-1", "document_keyword": "current-1.txt", "content": "Current rank one.", "similarity": 0.87}, {"dataset_id": DATASET_ID_2, "document_id": "current-2", "document_keyword": "current-2.txt", "content": "Current rank two.", "similarity": 0.86}, ], "doc_aggs": [ {"doc_id": "current-1", "doc_name": "current-1.txt", "count": 1}, {"doc_id": "current-2", "doc_name": "current-2.txt", "count": 1}, ], "total": 2, }, }, ) _install(monkeypatch, fake, config=_config(datasets=[DATASET_ID_1, DATASET_ID_2], page_size=3)) result = await ragflow_tools.knowledge_search("policy") assert [call[1]["dataset_ids"] for call in fake.retrieve_calls] == [[DATASET_ID_1], [DATASET_ID_2]] assert result.index("Legacy rank one.") < result.index("Current rank one.") < result.index("Legacy rank two.") assert "Current rank two." not in result assert "current-2.txt (1 chunk)" not in result assert "(score " not in result assert DATASET_ID_1 not in result assert DATASET_ID_2 not in result @pytest.mark.anyio async def test_all_dataset_scope_skips_empty_datasets_before_grouped_retrieval(monkeypatch: pytest.MonkeyPatch) -> None: fake = FakeRAGFlowClient( all_datasets=[ _dataset(DATASET_ID_1, "Empty legacy", embedding_model=EMBEDDING_V2, chunk_count=0), _dataset(DATASET_ID_2, "Current", embedding_model=EMBEDDING_V3, chunk_count=4), ], retrieval_by_dataset_ids={(DATASET_ID_2,): {"chunks": [{"dataset_id": DATASET_ID_2, "document_keyword": "guide.txt", "content": "Searchable."}], "doc_aggs": [], "total": 1}}, ) _install(monkeypatch, fake, config=_config(datasets=None)) result = await ragflow_tools.knowledge_search("searchable") assert [call[1]["dataset_ids"] for call in fake.retrieve_calls] == [[DATASET_ID_2]] assert "Searchable." in result @pytest.mark.anyio async def test_all_dataset_scope_skips_empty_dataset_without_embedding_model_and_warns( monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, ) -> None: fake = FakeRAGFlowClient( all_datasets=[ _dataset(DATASET_ID_1, "Unconfigured empty", embedding_model="", chunk_count=0), _dataset(DATASET_ID_2, "Current", embedding_model=EMBEDDING_V3, chunk_count=4), ], retrieval_by_dataset_ids={ (DATASET_ID_2,): { "chunks": [ { "dataset_id": DATASET_ID_2, "document_keyword": "guide.txt", "content": "Remaining dataset result.", "similarity": 0.75, } ], "doc_aggs": [], "total": 1, } }, ) _install(monkeypatch, fake, config=_config(datasets=None)) with caplog.at_level(logging.WARNING, logger="deerflow.community.ragflow.tools"): result = await ragflow_tools.knowledge_search("searchable") assert [call[1]["dataset_ids"] for call in fake.retrieve_calls] == [[DATASET_ID_2]] assert "Remaining dataset result." in result assert "(score 0.75)" in result assert DATASET_ID_1 not in result assert "Skipping empty RAGFlow dataset without embedding model metadata" in caplog.text assert DATASET_ID_1 in caplog.text @pytest.mark.anyio async def test_all_empty_dataset_scope_returns_no_content_without_retrieval(monkeypatch: pytest.MonkeyPatch) -> None: fake = FakeRAGFlowClient( all_datasets=[ _dataset(DATASET_ID_1, "Empty legacy", embedding_model=EMBEDDING_V2, chunk_count=0), _dataset(DATASET_ID_2, "Empty current", embedding_model=EMBEDDING_V3, chunk_count=0), ] ) _install(monkeypatch, fake, config=_config(datasets=None)) result = await ragflow_tools.knowledge_search("anything") assert result == "No relevant content found." assert fake.retrieve_calls == [] @pytest.mark.anyio async def test_grouped_retrieval_limits_concurrency_to_four(monkeypatch: pytest.MonkeyPatch) -> None: class ConcurrencyTrackingClient(FakeRAGFlowClient): def __init__(self) -> None: super().__init__(all_datasets=[_dataset(f"dataset-{index}", f"Dataset {index}", embedding_model=f"embedding-{index}@provider") for index in range(5)]) self.active_retrievals = 0 self.max_active_retrievals = 0 async def retrieve(self, query: str, **kwargs: object) -> dict: self.retrieve_calls.append((query, kwargs)) self.active_retrievals += 1 self.max_active_retrievals = max(self.max_active_retrievals, self.active_retrievals) try: await asyncio.sleep(0.05) return {"chunks": [], "doc_aggs": [], "total": 0} finally: self.active_retrievals -= 1 fake = ConcurrencyTrackingClient() _install(monkeypatch, fake, config=_config(datasets=None)) result = await ragflow_tools.knowledge_search("anything") assert result == "No relevant content found." assert len(fake.retrieve_calls) == 5 assert fake.max_active_retrievals == 4 @pytest.mark.anyio async def test_selected_scope_dataset_validation_is_bounded_and_parallel(monkeypatch: pytest.MonkeyPatch) -> None: dataset_ids = [f"dataset-{index}" for index in range(5)] class ConcurrencyTrackingClient(FakeRAGFlowClient): def __init__(self) -> None: super().__init__(datasets_by_id={dataset_id: [_dataset(dataset_id, f"Dataset {index}")] for index, dataset_id in enumerate(dataset_ids)}) self.active_dataset_lists = 0 self.max_active_dataset_lists = 0 async def list_datasets(self, *, dataset_id: str | None = None) -> list[dict]: self.list_calls.append(dataset_id) self.active_dataset_lists += 1 self.max_active_dataset_lists = max( self.max_active_dataset_lists, self.active_dataset_lists, ) try: await asyncio.sleep(0.05) return self.datasets_by_id.get(dataset_id or "", []) finally: self.active_dataset_lists -= 1 fake = ConcurrencyTrackingClient() _install(monkeypatch, fake, config=_config(datasets=dataset_ids)) result = await ragflow_tools.knowledge_search( "anything", knowledge_scope={ "version": 1, "mode": "selected", "dataset_ids": dataset_ids, }, ) assert result == "No relevant content found." assert fake.max_active_dataset_lists == 4 @pytest.mark.anyio async def test_selected_scope_document_validation_is_bounded_and_parallel(monkeypatch: pytest.MonkeyPatch) -> None: dataset_ids = [f"dataset-{index}" for index in range(5)] document_ids = [f"document-{index}" for index in range(5)] class ConcurrencyTrackingClient(FakeRAGFlowClient): def __init__(self) -> None: super().__init__( datasets_by_id={dataset_id: [_dataset(dataset_id, f"Dataset {index}")] for index, dataset_id in enumerate(dataset_ids)}, documents_by_dataset_id={ dataset_id: [ { "id": document_ids[index], "name": f"Document {index}", "run": "DONE", "chunk_count": 1, } ] for index, dataset_id in enumerate(dataset_ids) }, ) self.active_document_lists = 0 self.max_active_document_lists = 0 async def list_documents( self, dataset_id: str, *, params: list[tuple[str, str]], ) -> dict: self.document_list_calls.append((dataset_id, params)) self.active_document_lists += 1 self.max_active_document_lists = max( self.max_active_document_lists, self.active_document_lists, ) try: await asyncio.sleep(0.05) documents = self.documents_by_dataset_id[dataset_id] return { "code": 0, "data": {"docs": documents, "total": len(documents)}, } finally: self.active_document_lists -= 1 fake = ConcurrencyTrackingClient() _install(monkeypatch, fake, config=_config(datasets=dataset_ids)) result = await ragflow_tools.knowledge_search( "anything", knowledge_scope={ "version": 1, "mode": "selected", "dataset_ids": dataset_ids, "document_filters": [ { "dataset_id": dataset_id, "document_ids": [document_ids[index]], } for index, dataset_id in enumerate(dataset_ids) ], }, ) assert result == "No relevant content found." assert fake.max_active_document_lists == 4 @pytest.mark.anyio async def test_dataset_without_embedding_metadata_returns_protocol_error(monkeypatch: pytest.MonkeyPatch) -> None: fake = FakeRAGFlowClient(all_datasets=[{"id": DATASET_ID_1, "name": "Broken", "chunk_count": 1}]) _install(monkeypatch, fake, config=_config(datasets=None)) result = await ragflow_tools.knowledge_search("anything") assert result == "Error: RAGFlow request failed: RAGFlow returned a searchable dataset without embedding model metadata." assert fake.retrieve_calls == [] @pytest.mark.anyio async def test_group_failure_remains_strict_and_redacts_secret_and_dataset_id(monkeypatch: pytest.MonkeyPatch) -> None: fake = FakeRAGFlowClient( all_datasets=[ _dataset(DATASET_ID_1, "Legacy", embedding_model=EMBEDDING_V2), _dataset(DATASET_ID_2, "Current", embedding_model=EMBEDDING_V3), ], retrieval_by_dataset_ids={(DATASET_ID_2,): {"chunks": [], "doc_aggs": [], "total": 0}}, retrieval_errors_by_dataset_ids={(DATASET_ID_1,): RAGFlowAPIError(f"dataset {DATASET_ID_1} rejected ragflow-secret", code=102)}, ) _install(monkeypatch, fake, config=_config(datasets=None)) result = await ragflow_tools.knowledge_search("anything") assert result == "Error: dataset [DATASET_ID] rejected [REDACTED]" assert len(fake.retrieve_calls) == 2 @pytest.mark.anyio async def test_selected_scope_intersects_operator_allowlist_and_fails_closed( monkeypatch: pytest.MonkeyPatch, ) -> None: fake = FakeRAGFlowClient( datasets_by_id={ DATASET_ID_1: [_dataset(DATASET_ID_1, "Allowed")], DATASET_ID_2: [_dataset(DATASET_ID_2, "Not allowed")], }, ) _install(monkeypatch, fake, config=_config(datasets=[DATASET_ID_1])) result = await ragflow_tools.knowledge_search( "leave", knowledge_scope={ "version": 1, "mode": "selected", "dataset_ids": [DATASET_ID_2], }, ) assert result == ("Error: The selected knowledge scope is no longer available; choose the knowledge bases again.") assert fake.list_calls == [] assert fake.retrieve_calls == [] @pytest.mark.anyio async def test_selected_document_scope_validates_membership_and_splits_groups( monkeypatch: pytest.MonkeyPatch, ) -> None: fake = FakeRAGFlowClient( datasets_by_id={ DATASET_ID_1: [_dataset(DATASET_ID_1, "All files")], DATASET_ID_2: [_dataset(DATASET_ID_2, "One file")], }, documents_by_dataset_id={ DATASET_ID_2: [ { "id": "doc-1", "name": "guide.pdf", "run": "DONE", "chunk_count": 3, } ] }, ) _install( monkeypatch, fake, config=_config(datasets=[DATASET_ID_1, DATASET_ID_2]), ) result = await ragflow_tools.knowledge_search( "rice", knowledge_scope={ "version": 1, "mode": "selected", "dataset_ids": [DATASET_ID_1, DATASET_ID_2], "document_filters": [{"dataset_id": DATASET_ID_2, "document_ids": ["doc-1"]}], }, ) assert result == "No relevant content found." assert [call[1] for call in fake.retrieve_calls] == [ { "dataset_ids": [DATASET_ID_1], "page_size": 8, "similarity_threshold": 0.2, "vector_similarity_weight": 0.3, "top_k": 256, }, { "dataset_ids": [DATASET_ID_2], "document_ids": ["doc-1"], "page_size": 8, "similarity_threshold": 0.2, "vector_similarity_weight": 0.3, "top_k": 256, }, ] assert fake.document_list_calls == [ ( DATASET_ID_2, [ ("page", "1"), ("page_size", "1"), ("ids", "doc-1"), ], ) ] @pytest.mark.anyio @pytest.mark.parametrize( "documents", [ [], [{"id": "other", "run": "DONE", "chunk_count": 1}], [{"id": "doc-1", "run": "RUNNING", "chunk_count": 0}], [{"id": "doc-1", "run": "DONE", "chunk_count": 0}], ], ) async def test_invalid_or_unsearchable_document_selection_fails_closed( monkeypatch: pytest.MonkeyPatch, documents: list[dict], ) -> None: fake = FakeRAGFlowClient( datasets_by_id={ DATASET_ID_1: [_dataset(DATASET_ID_1, "Selected")], }, documents_by_dataset_id={DATASET_ID_1: documents}, ) _install(monkeypatch, fake) result = await ragflow_tools.knowledge_search( "rice", knowledge_scope={ "version": 1, "mode": "selected", "dataset_ids": [DATASET_ID_1], "document_filters": [{"dataset_id": DATASET_ID_1, "document_ids": ["doc-1"]}], }, ) assert result == ("Error: The selected knowledge scope is no longer available; choose the knowledge bases or files again.") assert fake.retrieve_calls == [] @pytest.mark.anyio async def test_disabled_scope_rejects_direct_tool_execution( monkeypatch: pytest.MonkeyPatch, ) -> None: fake = FakeRAGFlowClient() _install(monkeypatch, fake) result = await ragflow_tools.knowledge_search( "rice", knowledge_scope={"version": 1, "mode": "disabled"}, ) assert result == "Error: Knowledge search is disabled for this turn." assert fake.list_calls == [] assert fake.retrieve_calls == [] @pytest.mark.anyio async def test_missing_dataset_binding_with_empty_catalog_returns_guidance(monkeypatch: pytest.MonkeyPatch) -> None: fake = FakeRAGFlowClient() _install(monkeypatch, fake, config=_config(datasets=None)) result = await ragflow_tools.knowledge_search("leave") assert result == "Error: No accessible RAGFlow datasets were found; configure knowledge_search.datasets or add a dataset in RAGFlow." assert fake.list_calls == [None] assert fake.retrieve_calls == [] @pytest.mark.anyio async def test_missing_api_key_returns_english_guidance_and_warns_only_once( monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, ) -> None: fake = FakeRAGFlowClient() _install(monkeypatch, fake, config=_config(api_key=None, datasets=[DATASET_ID_1])) with caplog.at_level(logging.WARNING, logger="deerflow.community.ragflow.tools"): first = await ragflow_tools.knowledge_search("leave") second = await ragflow_tools.knowledge_search("benefits") assert first == "Error: RAGFlow API key is not configured; set knowledge_search.api_key in config.yaml (prefer $RAGFLOW_API_KEY)." assert second == first assert caplog.text.count("RAGFlow API key is not configured") == 1 assert fake.list_calls == [] @pytest.mark.anyio async def test_missing_knowledge_search_config_returns_english_guidance(monkeypatch: pytest.MonkeyPatch) -> None: fake = FakeRAGFlowClient() _install(monkeypatch, fake, config=_config(configured=False, datasets=[DATASET_ID_1])) result = await ragflow_tools.knowledge_search("leave") assert result == "Error: knowledge_search is not configured; add its RAGFlow settings to the tools list in config.yaml." assert fake.list_calls == [] @pytest.mark.anyio async def test_api_error_is_returned_as_readable_text(monkeypatch: pytest.MonkeyPatch) -> None: fake = FakeRAGFlowClient( datasets_by_id={DATASET_ID_1: [_dataset(DATASET_ID_1, "Policies")]}, retrieval_errors_by_dataset_ids={(DATASET_ID_1,): RAGFlowAPIError("embedding models do not match", code=102)}, ) _install(monkeypatch, fake) result = await ragflow_tools.knowledge_search("leave") assert result == "Error: embedding models do not match" @pytest.mark.anyio async def test_error_path_redacts_dataset_uuid(monkeypatch: pytest.MonkeyPatch) -> None: dataset_id = "0123456789abcdef0123456789abcdef" fake = FakeRAGFlowClient( datasets_by_id={DATASET_ID_1: [_dataset(DATASET_ID_1, "Policies")]}, retrieval_errors_by_dataset_ids={(DATASET_ID_1,): RAGFlowAPIError(f"dataset {dataset_id} failed", code=102)}, ) _install(monkeypatch, fake) result = await ragflow_tools.knowledge_search("leave") assert dataset_id not in result assert "[DATASET_ID]" in result @pytest.mark.anyio async def test_success_path_preserves_legitimate_uuid_and_md5_text(monkeypatch: pytest.MonkeyPatch) -> None: uuid = "123e4567-e89b-12d3-a456-426614174000" md5 = "d41d8cd98f00b204e9800998ecf8427e" fake = FakeRAGFlowClient( datasets_by_id={DATASET_ID_1: [_dataset(DATASET_ID_1, "HR Policies")]}, retrieval={ "chunks": [ { "dataset_id": DATASET_ID_1, "document_keyword": "checksums.txt", "content": f"Trace {uuid}; checksum {md5}.", } ] }, ) _install(monkeypatch, fake) result = await ragflow_tools.knowledge_search("trace") assert uuid in result assert md5 in result assert "[DATASET_ID]" not in result @pytest.mark.anyio async def test_success_path_still_redacts_api_key(monkeypatch: pytest.MonkeyPatch) -> None: fake = FakeRAGFlowClient( datasets_by_id={DATASET_ID_1: [_dataset(DATASET_ID_1, "HR Policies")]}, retrieval={ "chunks": [ { "dataset_id": DATASET_ID_1, "document_keyword": "secret.txt", "content": "Accidental echo: ragflow-secret", } ] }, ) _install(monkeypatch, fake) result = await ragflow_tools.knowledge_search("secret") assert "ragflow-secret" not in result assert "[REDACTED]" in result @pytest.mark.anyio async def test_connection_error_is_english_and_does_not_leak_key( monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, ) -> None: fake = FakeRAGFlowClient(error=RAGFlowConnectionError("ConnectError: refused ragflow-secret")) _install(monkeypatch, fake) with caplog.at_level(logging.WARNING, logger="deerflow.community.ragflow.tools"): result = await ragflow_tools.knowledge_search("leave") assert result == "Error: Unable to connect to RAGFlow (http://ragflow.test): ConnectError: refused [REDACTED]" assert "ragflow-secret" not in result assert "ragflow-secret" not in caplog.text @pytest.mark.anyio @pytest.mark.parametrize( "base_url", [ "http://ragflow-secret@ragflow.test", "http://ragflow%2Dsecret@ragflow.test", "http://user:ragflow-secret@ragflow.test", ], ) async def test_base_url_with_plain_or_encoded_userinfo_is_rejected_without_leaking_credentials( monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, base_url: str, ) -> None: fake = FakeRAGFlowClient() _install(monkeypatch, fake, config=_config(base_url=base_url, datasets=[DATASET_ID_1])) with caplog.at_level(logging.WARNING, logger="deerflow.community.ragflow.tools"): result = await ragflow_tools.knowledge_search("leave") assert result == "Error: Invalid RAGFlow settings for knowledge_search; check config.yaml." assert "ragflow-secret" not in result assert "ragflow-secret" not in caplog.text assert "ragflow%2Dsecret" not in caplog.text assert fake.list_calls == [] @pytest.mark.anyio async def test_empty_query_has_english_error(monkeypatch: pytest.MonkeyPatch) -> None: fake = FakeRAGFlowClient() _install(monkeypatch, fake) result = await ragflow_tools.knowledge_search(" ") assert result == "Error: query must not be empty." assert fake.list_calls == [] @pytest.mark.anyio async def test_empty_retrieval_has_explicit_english_message(monkeypatch: pytest.MonkeyPatch) -> None: fake = FakeRAGFlowClient(datasets_by_id={DATASET_ID_1: [_dataset(DATASET_ID_1, "HR Policies")]}) _install(monkeypatch, fake) result = await ragflow_tools.knowledge_search("nothing") assert result == "No relevant content found." def test_formatting_uses_only_normalized_chunk_fields() -> None: result = format_retrieval_result( { "chunks": [ { "kb_id": "dataset-1", "doc_id": "doc-legacy", "docnm_kwd": "legacy.pdf", "content": "abcdefghij", "similarity": 0.5, } ] }, dataset_names_by_id={"dataset-1": "HR Policies"}, max_chars_per_chunk=5, max_total_chars=1000, ) assert "Unknown dataset / Unknown document" in result assert "HR Policies" not in result assert "legacy.pdf" not in result assert "abcd…" in result assert "abcdefghij" not in result assert "dataset-1" not in result def test_formatting_applies_total_response_truncation_in_english() -> None: result = format_retrieval_result( { "chunks": [ { "dataset_id": "dataset-1", "document_keyword": f"document-{index}.txt", "content": "content " * 20, "similarity": 0.5, } for index in range(4) ] }, dataset_names_by_id={"dataset-1": "Policies"}, max_chars_per_chunk=100, max_total_chars=120, ) assert len(result) <= 120 assert result.endswith("… (response truncated)") def test_retrieval_settings_load_bound_dataset_ids_and_hide_secret(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(ragflow_tools, "get_app_config", lambda: _config(datasets=[DATASET_ID_1, DATASET_ID_2])) config, error = ragflow_tools._settings_or_error() assert error is None assert config is not None assert config.datasets == [DATASET_ID_1, DATASET_ID_2] assert str(config.base_url).rstrip("/") == "http://ragflow.test" assert config.page_size == 8 assert config.max_chars_per_chunk == 800 assert config.max_total_chars == 8000 assert "ragflow-secret" not in repr(config) def test_retrieval_settings_allow_omitting_dataset_ids(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(ragflow_tools, "get_app_config", lambda: _config(datasets=None)) config, error = ragflow_tools._settings_or_error() assert error is None assert config is not None assert config.datasets is None def test_retrieval_settings_do_not_fall_back_to_knowledge_capability_config( monkeypatch: pytest.MonkeyPatch, ) -> None: legacy_global = SimpleNamespace( enabled=True, base_url="http://legacy-ragflow.test", api_key="legacy-secret", timeout=60, ) app_config = SimpleNamespace( knowledge_base=legacy_global, get_tool_config=lambda _name: None, ) monkeypatch.setattr(ragflow_tools, "get_app_config", lambda: app_config) settings, error = ragflow_tools._settings_or_error() assert settings is None assert error == "Error: knowledge_search is not configured; add its RAGFlow settings to the tools list in config.yaml." @pytest.mark.anyio async def test_explicitly_empty_dataset_allowlist_fails_closed(monkeypatch: pytest.MonkeyPatch) -> None: fake = FakeRAGFlowClient(all_datasets=[_dataset(DATASET_ID_1, "Must remain inaccessible")]) _install(monkeypatch, fake, config=_config(datasets=[])) result = await ragflow_tools.knowledge_search("leave") assert result == "Error: Invalid RAGFlow settings for knowledge_search; check config.yaml." assert fake.list_calls == [] assert fake.retrieve_calls == [] def test_agent_exposes_search_and_name_listing_tools() -> None: assert hasattr(ragflow_tools, "list_knowledge_bases_tool") assert hasattr(ragflow_tools, "list_knowledge_bases") assert ragflow_tools.knowledge_search_tool.name == "knowledge_search" assert ragflow_tools.knowledge_search_tool.coroutine is not None assert set(ragflow_tools.knowledge_search_tool.tool_call_schema.model_fields) == {"query"} assert ragflow_tools.list_knowledge_bases_tool.name == "list_knowledge_bases" assert ragflow_tools.list_knowledge_bases_tool.coroutine is not None assert not ragflow_tools.list_knowledge_bases_tool.tool_call_schema.model_fields def test_tool_assembly_hides_bound_dataset_ids_without_network_io(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(ragflow_tools, "_build_client", lambda settings: pytest.fail("tool assembly must not perform network IO")) tool_config = ToolConfig( name="knowledge_search", group="knowledge", use="deerflow.community.ragflow.tools:knowledge_search_tool", base_url="http://ragflow.test", api_key="ragflow-secret", datasets=[DATASET_ID_1, DATASET_ID_2], ) config = SimpleNamespace( tools=[tool_config], knowledge_base=SimpleNamespace(enabled=True), sandbox=SimpleNamespace(use="example.remote:Sandbox"), skill_evolution=SimpleNamespace(enabled=False), models=[], acp_agents={}, get_model_config=lambda name: None, ) tools = get_available_tools(include_mcp=False, app_config=config) assembled = next(tool for tool in tools if tool.name == "knowledge_search") assert "If knowledge_search.datasets is omitted" in assembled.description assert "all datasets accessible to the configured RAGFlow API key" in assembled.description assert DATASET_ID_1 not in assembled.description assert DATASET_ID_2 not in assembled.description assert "ragflow-secret" not in assembled.description assert {tool.name for tool in tools}.isdisjoint({"list_knowledge_bases"}) def test_tool_assembly_hides_configured_knowledge_provider_when_capability_is_disabled() -> None: tool_config = ToolConfig( name="knowledge_search", group="knowledge", use="deerflow.community.ragflow.tools:knowledge_search_tool", base_url="http://ragflow.test", api_key="ragflow-secret", ) config = SimpleNamespace( tools=[tool_config], knowledge_base=SimpleNamespace(enabled=False), sandbox=SimpleNamespace(use="example.remote:Sandbox"), skill_evolution=SimpleNamespace(enabled=False), models=[], acp_agents={}, get_model_config=lambda _name: None, ) tools = get_available_tools(include_mcp=False, app_config=config) assert {tool.name for tool in tools}.isdisjoint({"knowledge_search"}) def test_ragflow_package_has_explicit_init_file() -> None: package_dir = Path(ragflow_tools.__file__).resolve().parent assert (package_dir / "__init__.py").is_file() @pytest.mark.asyncio async def test_search_artifact_binds_citation_to_exact_retrieved_document(monkeypatch: pytest.MonkeyPatch) -> None: client = FakeRAGFlowClient( all_datasets=[_dataset(DATASET_ID_1, "Engineering")], retrieval={"chunks": [{"id": "chunk-a", "dataset_id": DATASET_ID_1, "document_id": "doc-a", "document_keyword": "Manual.pdf", "content": "The limit is 42.", "positions": [[3, 10, 20, 30, 40]]}]}, ) monkeypatch.setattr(ragflow_tools, "get_app_config", lambda: _config()) monkeypatch.setattr(ragflow_tools, "_build_client", lambda _: client) runtime = SimpleNamespace(context={}) content, artifact = await ragflow_tools._knowledge_search_entrypoint("limit", runtime) source = artifact["knowledge_sources"]["sources"][0] assert source["document_id"] == "doc-a" assert source["chunk_id"] == "chunk-a" assert source["document_name"] == "Manual.pdf" assert source["text"] == "The limit is 42." assert source["pages"] == [3] assert f"](#knowledge-{source['id']})" in content assert DATASET_ID_1 not in content assert "doc-a" not in content _, next_artifact = await ragflow_tools._knowledge_search_entrypoint("limit", runtime) assert next_artifact["knowledge_sources"]["sources"][0]["id"] != source["id"] @pytest.mark.asyncio async def test_search_artifact_redacts_credentials_and_has_no_sources_on_error(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(ragflow_tools, "get_app_config", lambda: _config()) client = FakeRAGFlowClient( all_datasets=[_dataset(DATASET_ID_1, "ragflow-secret")], retrieval={"chunks": [{"id": "chunk-a", "dataset_id": DATASET_ID_1, "document_id": "doc-a", "document_keyword": "ragflow-secret", "content": "ragflow-secret"}]}, ) monkeypatch.setattr(ragflow_tools, "_build_client", lambda _: client) result = await ragflow_tools._knowledge_search_entrypoint("query", SimpleNamespace(context={})) assert "ragflow-secret" not in str(result) content, artifact = await ragflow_tools._knowledge_search_entrypoint("", SimpleNamespace(context={})) assert content.startswith("Error:") assert artifact is None @pytest.mark.asyncio async def test_citation_artifact_survives_native_tool_node_and_message_serialization(monkeypatch: pytest.MonkeyPatch) -> None: from langchain_core.messages import AIMessage, messages_from_dict, messages_to_dict from langgraph.graph import END, START, MessagesState, StateGraph from langgraph.prebuilt import ToolNode client = FakeRAGFlowClient( all_datasets=[_dataset(DATASET_ID_1, "Engineering")], retrieval={"chunks": [{"id": "chunk-a", "dataset_id": DATASET_ID_1, "document_id": "doc-a", "document_keyword": "Manual.pdf", "content": "Limit: 42."}]}, ) monkeypatch.setattr(ragflow_tools, "get_app_config", lambda: _config()) monkeypatch.setattr(ragflow_tools, "_build_client", lambda _: client) builder = StateGraph(MessagesState) builder.add_node("tools", ToolNode([ragflow_tools.knowledge_search_tool])) builder.add_edge(START, "tools") builder.add_edge("tools", END) result = await builder.compile().ainvoke({"messages": [AIMessage(content="", tool_calls=[{"name": "knowledge_search", "args": {"query": "limit"}, "id": "call-a"}])]}) tool_message = result["messages"][-1] assert tool_message.status == "success" assert tool_message.artifact["knowledge_sources"]["sources"][0]["document_id"] == "doc-a" restored = messages_from_dict(messages_to_dict([tool_message]))[0] assert restored.artifact == tool_message.artifact assert restored.content == tool_message.content def test_subagent_forwards_only_cited_captured_sources() -> None: from deerflow.community.ragflow.sources import cited_source_artifact sources = [{"id": "a", "text": "first"}, {"id": "b", "text": "second"}] messages = [{"type": "tool", "name": "knowledge_search", "artifact": {"knowledge_sources": {"version": 1, "sources": sources}}}] assert cited_source_artifact(messages, "[citation:1](#knowledge-a)") == {"knowledge_sources": {"version": 1, "sources": [sources[0]]}} assert cited_source_artifact(messages, "[citation:3](#knowledge-invented)") is None assert cited_source_artifact([{**messages[0], "type": "ai"}], "[citation:1](#knowledge-a)") is None def test_citation_budget_never_emits_partial_links_or_unseen_artifact_text() -> None: from deerflow.community.ragflow.formatting import format_retrieval_sources chunks = [{"id": f"chunk-{i}", "dataset_id": DATASET_ID_1, "document_id": "doc", "content": "X" * 1000} for i in range(4)] content, artifact = format_retrieval_sources({"chunks": chunks}, dataset_names_by_id={DATASET_ID_1: "Knowledge"}, max_total_chars=200, max_chars_per_chunk=100) assert len(content) <= 200 sources = artifact["knowledge_sources"]["sources"] assert len(sources) == 1 assert f"](#knowledge-{sources[0]['id']})" in content assert sources[0]["text"] in content assert sources[0]["truncated"] is True def test_task_command_preserves_child_source_artifact() -> None: from deerflow.tools.builtins.task_tool import _task_result_command source = {"id": "abc", "text": "evidence"} message = {"type": "tool", "name": "knowledge_search", "artifact": {"knowledge_sources": {"version": 1, "sources": [source]}}} command = _task_result_command(tool_call_id="task-1", status="completed", result="Answer [citation:1](#knowledge-abc)", source_messages=[message]) assert command.update["messages"][0].artifact["knowledge_sources"]["sources"] == [source]