test: use concrete knowledge node adapters and source records (#43211)

api/tests/unit_tests/core/workflow/nodes/knowledge_index/test_knowledge_index_node.py

@@ -1,6 +1,5 @@
 import time
 import uuid
-from unittest.mock import Mock
 
 import pytest
 from pytest_mock import MockerFixture
@@ -9,14 +8,14 @@ from sqlalchemy.orm import Session
 
 from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom
 from core.rag.index_processor.constant.index_type import IndexTechniqueType
+from core.rag.index_processor.index_processor import IndexProcessor
+from core.rag.summary_index.summary_index import SummaryIndex
 from core.workflow.nodes.knowledge_index.entities import KnowledgeIndexNodeData
 from core.workflow.nodes.knowledge_index.exc import KnowledgeIndexNodeError
 from core.workflow.nodes.knowledge_index.knowledge_index_node import KnowledgeIndexNode
 from core.workflow.nodes.knowledge_index.protocols import (
-    IndexProcessorProtocol,
     Preview,
     PreviewItem,
-    SummaryIndexServiceProtocol,
 )
 from core.workflow.system_variables import SystemVariableKey, build_system_variables
 from graphon.enums import WorkflowNodeExecutionStatus
@@ -54,8 +53,10 @@ def mock_graph_runtime_state():
 
 @pytest.fixture
 def mock_index_processor(mocker: MockerFixture):
-    """Create mock IndexProcessorProtocol."""
-    mock_processor = Mock(spec=IndexProcessorProtocol)
+    """Use the real processor with indexing and preview I/O boundaries patched."""
+    mock_processor = IndexProcessor()
+    mocker.patch.object(mock_processor, "get_preview_output")
+    mocker.patch.object(mock_processor, "index_and_clean")
     mocker.patch(
         "core.workflow.nodes.knowledge_index.knowledge_index_node.IndexProcessor",
         return_value=mock_processor,
@@ -65,8 +66,9 @@ def mock_index_processor(mocker: MockerFixture):
 
 @pytest.fixture
 def mock_summary_index_service(mocker: MockerFixture):
-    """Create mock SummaryIndexServiceProtocol."""
-    mock_service = Mock(spec=SummaryIndexServiceProtocol)
+    """Use the real summary service with generation patched."""
+    mock_service = SummaryIndex()
+    mocker.patch.object(mock_service, "generate_and_vectorize_summary")
     mocker.patch(
         "core.workflow.nodes.knowledge_index.knowledge_index_node.SummaryIndex",
         return_value=mock_service,

api/tests/unit_tests/core/workflow/nodes/knowledge_retrieval/test_knowledge_retrieval_node.py

@@ -6,6 +6,7 @@ import pytest
 from pytest_mock import MockerFixture
 
 from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom
+from core.rag.retrieval.dataset_retrieval import DatasetRetrieval
 from core.workflow.nodes.knowledge_retrieval.entities import (
     Condition,
     KnowledgeRetrievalNodeData,
@@ -20,15 +21,29 @@ from core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node import (
     _normalize_metadata_filter_scalar,
     _normalize_metadata_filter_sequence_item,
 )
-from core.workflow.nodes.knowledge_retrieval.retrieval import RAGRetrievalProtocol, Source
+from core.workflow.nodes.knowledge_retrieval.retrieval import Source, SourceMetadata
 from core.workflow.system_variables import build_system_variables
 from graphon.enums import WorkflowNodeExecutionStatus
 from graphon.model_runtime.entities.llm_entities import LLMUsage
 from graphon.runtime import GraphRuntimeState, VariablePool
-from graphon.variables import StringSegment
+from graphon.variables import ArrayObjectSegment, StringSegment
 from tests.workflow_test_utils import build_test_graph_init_params
 
 
+def _source() -> Source:
+    return Source(
+        metadata=SourceMetadata(
+            dataset_id="dataset-id",
+            dataset_name="Knowledge",
+            document_id="document-id",
+            document_name="Python",
+            data_source_type="upload_file",
+        ),
+        title="Python",
+        content="Python is a programming language",
+    )
+
+
 @pytest.fixture
 def mock_graph_init_params():
     """Create mock GraphInitParams."""
@@ -58,10 +73,9 @@ def mock_graph_runtime_state():
 
 @pytest.fixture
 def mock_rag_retrieval(mocker: MockerFixture):
-    """Create mock RAGRetrievalProtocol."""
-    mock_retrieval = Mock(spec=RAGRetrievalProtocol)
-    mock_retrieval.knowledge_retrieval.return_value = []
-    mock_retrieval.llm_usage = LLMUsage.empty_usage()
+    """Keep real usage and request metadata while isolating retrieval I/O."""
+    mock_retrieval = DatasetRetrieval()
+    mocker.patch.object(mock_retrieval, "knowledge_retrieval", return_value=[])
     mocker.patch(
         "core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node.DatasetRetrieval",
         return_value=mock_retrieval,
@@ -173,10 +187,8 @@ class TestKnowledgeRetrievalNode:
         }
 
         # Mock retrieval response
-        mock_source = Mock(spec=Source)
-        mock_source.model_dump.return_value = {"content": "Python is a programming language"}
+        mock_source = _source()
         mock_rag_retrieval.knowledge_retrieval.return_value = [mock_source]
-        mock_rag_retrieval.llm_usage = LLMUsage.empty_usage()
 
         node = KnowledgeRetrievalNode(
             node_id=node_id,
@@ -192,7 +204,7 @@ class TestKnowledgeRetrievalNode:
         assert result.status == WorkflowNodeExecutionStatus.SUCCEEDED
         assert "result" in result.outputs
         assert mock_rag_retrieval.knowledge_retrieval.called
-        mock_source.model_dump.assert_called_once_with(by_alias=True)
+        assert result.outputs["result"] == ArrayObjectSegment(value=[mock_source.model_dump(by_alias=True)])
 
     def test_run_with_query_variable_multiple_mode(
         self,
@@ -217,10 +229,8 @@ class TestKnowledgeRetrievalNode:
         }
 
         # Mock retrieval response
-        mock_source = Mock(spec=Source)
-        mock_source.model_dump.return_value = {"content": "Python is a programming language"}
+        mock_source = _source()
         mock_rag_retrieval.knowledge_retrieval.return_value = [mock_source]
-        mock_rag_retrieval.llm_usage = LLMUsage.empty_usage()
 
         node = KnowledgeRetrievalNode(
             node_id=node_id,
@@ -332,7 +342,6 @@ class TestKnowledgeRetrievalNode:
         mock_rag_retrieval.knowledge_retrieval.side_effect = RateLimitExceededError(
             "knowledge base request rate limit exceeded"
         )
-        mock_rag_retrieval.llm_usage = LLMUsage.empty_usage()
 
         node = KnowledgeRetrievalNode(
             node_id=node_id,
@@ -371,7 +380,6 @@ class TestKnowledgeRetrievalNode:
 
         # Mock retrieval to raise generic exception
         mock_rag_retrieval.knowledge_retrieval.side_effect = Exception("Unexpected error")
-        mock_rag_retrieval.llm_usage = LLMUsage.empty_usage()
 
         node = KnowledgeRetrievalNode(
             node_id=node_id,
@@ -447,9 +455,8 @@ class TestFetchDatasetRetriever:
         )
 
         # Mock retrieval response
-        mock_source = Mock(spec=Source)
+        mock_source = _source()
         mock_rag_retrieval.knowledge_retrieval.return_value = [mock_source]
-        mock_rag_retrieval.llm_usage = LLMUsage.empty_usage()
 
         node_id = str(uuid.uuid4())
         config = {"id": node_id, "data": node_data.model_dump()}
@@ -483,7 +490,6 @@ class TestFetchDatasetRetriever:
 
         # Mock retrieval response
         mock_rag_retrieval.knowledge_retrieval.return_value = []
-        mock_rag_retrieval.llm_usage = LLMUsage.empty_usage()
 
         node_id = str(uuid.uuid4())
         config = {
@@ -538,7 +544,6 @@ class TestFetchDatasetRetriever:
 
         # Mock retrieval response
         mock_rag_retrieval.knowledge_retrieval.return_value = []
-        mock_rag_retrieval.llm_usage = LLMUsage.empty_usage()
 
         node_id = str(uuid.uuid4())
         config = {
@@ -655,7 +660,6 @@ class TestFetchDatasetRetriever:
         )
 
         mock_rag_retrieval.knowledge_retrieval.return_value = []
-        mock_rag_retrieval.llm_usage = LLMUsage.empty_usage()
 
         # Act
         node._fetch_dataset_retriever(Mock(), node_data=node_data, variables=variables)