test: move pure integration cases to unit tests (#40705)

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

api/tests/integration_tests/core/workflow/nodes/datasource/test_datasource_node_integration.py

@@ -1,90 +0,0 @@
-from pytest_mock import MockerFixture
-
-from core.app.entities.app_invoke_entities import DIFY_RUN_CONTEXT_KEY
-from core.workflow.nodes.datasource.datasource_node import DatasourceNode
-from core.workflow.nodes.datasource.entities import DatasourceNodeData
-from graphon.enums import WorkflowNodeExecutionStatus
-from graphon.node_events import NodeRunResult, StreamCompletedEvent
-
-
-class _Seg:
-    def __init__(self, v):
-        self.value = v
-
-
-class _VarPool:
-    def __init__(self, data):
-        self.data = data
-
-    def get(self, path):
-        d = self.data
-        for k in path:
-            d = d[k]
-        return _Seg(d)
-
-    def add(self, *_a, **_k):
-        pass
-
-
-class _GS:
-    def __init__(self, vp):
-        self.variable_pool = vp
-
-
-class _GP:
-    workflow_id = "wf-1"
-    graph_config = {}
-    run_context = {
-        DIFY_RUN_CONTEXT_KEY: {
-            "tenant_id": "t1",
-            "app_id": "app-1",
-            "user_id": "u1",
-            "user_from": "account",
-            "invoke_from": "debugger",
-        }
-    }
-    call_depth = 0
-
-
-def test_node_integration_minimal_stream(mocker: MockerFixture):
-    sys_d = {
-        "sys": {
-            "datasource_type": "online_document",
-            "datasource_info": {"workspace_id": "w", "page": {"page_id": "pg", "type": "t"}, "credential_id": ""},
-        }
-    }
-    vp = _VarPool(sys_d)
-
-    class _Mgr:
-        @classmethod
-        def get_icon_url(cls, **_):
-            return "icon"
-
-        @classmethod
-        def stream_node_events(cls, **_):
-            yield from ()
-            yield StreamCompletedEvent(node_run_result=NodeRunResult(status=WorkflowNodeExecutionStatus.SUCCEEDED))
-
-        @classmethod
-        def get_upload_file_by_id(cls, **_):
-            raise AssertionError
-
-    mocker.patch("core.workflow.nodes.datasource.datasource_node.DatasourceManager", new=_Mgr)
-
-    node = DatasourceNode(
-        node_id="n",
-        data=DatasourceNodeData(
-            type="datasource",
-            version="1",
-            title="Datasource",
-            provider_type="plugin",
-            provider_name="p",
-            plugin_id="plug",
-            datasource_name="ds",
-        ),
-        graph_init_params=_GP(),
-        graph_runtime_state=_GS(vp),
-    )
-
-    out = list(node._run())
-    assert isinstance(out[-1], StreamCompletedEvent)

api/tests/integration_tests/storage/test_clickzetta_volume.py

@@ -103,66 +103,6 @@ class TestClickZettaVolumeStorage(unittest.TestCase):
         storage.delete(test_filename)
         assert not storage.exists(test_filename)
 
-    def test_config_validation(self):
-        """Test configuration validation."""
-        # Test missing required fields
-        with pytest.raises(ValueError):
-            ClickZettaVolumeConfig(
-                username="",  # Empty username should fail
-                password="pass",
-                instance="instance",
-            )
-
-        # Test invalid volume type
-        with pytest.raises(ValueError):
-            ClickZettaVolumeConfig(username="user", password="pass", instance="instance", volume_type="invalid_type")
-
-        # Test external volume without volume_name
-        with pytest.raises(ValueError):
-            ClickZettaVolumeConfig(
-                username="user",
-                password="pass",
-                instance="instance",
-                volume_type="external",
-                # Missing volume_name
-            )
-
-    def test_volume_path_generation(self):
-        """Test volume path generation for different types."""
-        storage = ClickZettaVolumeStorage(self.config)
-
-        # Test table volume path
-        path = storage._get_volume_path("test.txt", "12345")
-        assert path == "test_dataset_12345/test.txt"
-
-        # Test path with existing dataset_id prefix
-        path = storage._get_volume_path("12345/test.txt")
-        assert path == "12345/test.txt"
-
-        # Test user volume
-        storage._config.volume_type = "user"
-        path = storage._get_volume_path("test.txt")
-        assert path == "test.txt"
-
-    def test_sql_prefix_generation(self):
-        """Test SQL prefix generation for different volume types."""
-        storage = ClickZettaVolumeStorage(self.config)
-
-        # Test table volume SQL prefix
-        prefix = storage._get_volume_sql_prefix("12345")
-        assert prefix == "TABLE VOLUME test_dataset_12345"
-
-        # Test user volume SQL prefix
-        storage._config.volume_type = "user"
-        prefix = storage._get_volume_sql_prefix()
-        assert prefix == "USER VOLUME"
-
-        # Test external volume SQL prefix
-        storage._config.volume_type = "external"
-        storage._config.volume_name = "my_external_volume"
-        prefix = storage._get_volume_sql_prefix()
-        assert prefix == "VOLUME my_external_volume"
-
 
 if __name__ == "__main__":
     unittest.main()

api/tests/integration_tests/utils/child_class.py

@@ -1,7 +0,0 @@
-from tests.integration_tests.utils.parent_class import ParentClass
-
-
-class ChildClass(ParentClass):
-    def __init__(self, name: str):
-        super().__init__(name)
-        self.name = name

api/tests/integration_tests/utils/lazy_load_class.py

@@ -1,7 +0,0 @@
-from tests.integration_tests.utils.parent_class import ParentClass
-
-
-class LazyLoadChildClass(ParentClass):
-    def __init__(self, name: str):
-        super().__init__(name)
-        self.name = name

api/tests/integration_tests/utils/parent_class.py

@@ -1,6 +0,0 @@
-class ParentClass:
-    def __init__(self, name):
-        self.name = name
-
-    def get_name(self):
-        return self.name

api/tests/integration_tests/utils/test_module_import_helper.py

@@ -1,34 +0,0 @@
-import os
-
-from core.helper.module_import_helper import import_module_from_source, load_single_subclass_from_source
-from tests.integration_tests.utils.parent_class import ParentClass
-
-
-def test_loading_subclass_from_source():
-    current_path = os.getcwd()
-    module = load_single_subclass_from_source(
-        module_name="ChildClass", script_path=os.path.join(current_path, "child_class.py"), parent_type=ParentClass
-    )
-    assert module
-    assert module.__name__ == "ChildClass"
-
-
-def test_load_import_module_from_source():
-    current_path = os.getcwd()
-    module = import_module_from_source(
-        module_name="ChildClass", py_file_path=os.path.join(current_path, "child_class.py")
-    )
-    assert module
-    assert module.__name__ == "ChildClass"
-
-
-def test_lazy_loading_subclass_from_source():
-    current_path = os.getcwd()
-    clz = load_single_subclass_from_source(
-        module_name="LazyLoadChildClass",
-        script_path=os.path.join(current_path, "lazy_load_class.py"),
-        parent_type=ParentClass,
-        use_lazy_loader=True,
-    )
-    instance = clz("dify")
-    assert instance.get_name() == "dify"

api/tests/unit_tests/core/datasource/test_stream_node_events.py

@@ -1,3 +1,5 @@
+"""Unit coverage for datasource event-stream accumulation."""
+
 from collections.abc import Generator
 
 from pytest_mock import MockerFixture
@@ -21,7 +23,7 @@ def _gen_var_stream() -> Generator[DatasourceMessage, None, None]:
     )
 
 
-def test_stream_node_events_accumulates_variables(mocker: MockerFixture):
+def test_stream_node_events_accumulates_variables(mocker: MockerFixture) -> None:
     mocker.patch.object(DatasourceManager, "stream_online_results", return_value=_gen_var_stream())
     events = list(
         DatasourceManager.stream_node_events(

api/tests/unit_tests/core/workflow/nodes/datasource/test_datasource_node_stream.py

@@ -0,0 +1,72 @@
+"""Unit coverage for datasource node event streaming."""
+
+import time
+from collections.abc import Generator
+from typing import Never
+
+from pytest_mock import MockerFixture
+
+from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom
+from core.workflow.nodes.datasource.datasource_node import DatasourceNode
+from core.workflow.nodes.datasource.entities import DatasourceNodeData
+from core.workflow.system_variables import build_system_variables
+from graphon.enums import WorkflowNodeExecutionStatus
+from graphon.node_events import NodeRunResult, StreamCompletedEvent
+from graphon.runtime import GraphRuntimeState, VariablePool
+from tests.workflow_test_utils import build_test_graph_init_params
+
+
+def test_node_integration_minimal_stream(mocker: MockerFixture) -> None:
+    variable_pool = VariablePool.from_bootstrap(
+        system_variables=build_system_variables(
+            datasource_type="online_document",
+            datasource_info={
+                "workspace_id": "w",
+                "page": {"page_id": "pg", "type": "t"},
+                "credential_id": "",
+            },
+        )
+    )
+    graph_init_params = build_test_graph_init_params(
+        workflow_id="wf-1",
+        tenant_id="t1",
+        app_id="app-1",
+        user_id="u1",
+        user_from=UserFrom.ACCOUNT,
+        invoke_from=InvokeFrom.DEBUGGER,
+    )
+    graph_runtime_state = GraphRuntimeState(variable_pool=variable_pool, start_at=time.perf_counter())
+
+    class _Mgr:
+        @classmethod
+        def get_icon_url(cls, **_: object) -> str:
+            return "icon"
+
+        @classmethod
+        def stream_node_events(cls, **_: object) -> Generator[StreamCompletedEvent, None, None]:
+            yield from ()
+            yield StreamCompletedEvent(node_run_result=NodeRunResult(status=WorkflowNodeExecutionStatus.SUCCEEDED))
+
+        @classmethod
+        def get_upload_file_by_id(cls, **_: object) -> Never:
+            raise AssertionError
+
+    mocker.patch("core.workflow.nodes.datasource.datasource_node.DatasourceManager", new=_Mgr)
+
+    node = DatasourceNode(
+        node_id="n",
+        data=DatasourceNodeData(
+            type="datasource",
+            version="1",
+            title="Datasource",
+            provider_type="plugin",
+            provider_name="p",
+            plugin_id="plug",
+            datasource_name="ds",
+        ),
+        graph_init_params=graph_init_params,
+        graph_runtime_state=graph_runtime_state,
+    )
+
+    out = list(node._run())
+    assert isinstance(out[-1], StreamCompletedEvent)

api/tests/unit_tests/core/workflow/nodes/template_transform/test_template_transform_execution.py

@@ -1,5 +1,8 @@
+"""Unit coverage for in-process template-transform execution."""
+
 import time
 import uuid
+from typing import cast
 
 from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom
 from core.workflow.node_factory import DifyNodeFactory
@@ -9,7 +12,7 @@ from graphon.graph import Graph
 from graphon.nodes.template_transform.entities import TemplateTransformNodeData
 from graphon.nodes.template_transform.template_transform_node import TemplateTransformNode
 from graphon.runtime import GraphRuntimeState, VariablePool
-from graphon.template_rendering import TemplateRenderError
+from graphon.template_rendering import Jinja2TemplateRenderer, TemplateRenderError
 from tests.workflow_test_utils import build_test_graph_init_params
 
 
@@ -26,7 +29,7 @@ class _SimpleJinja2Renderer:
             raise TemplateRenderError(str(exc)) from exc
 
 
-def test_execute_template_transform():
+def test_execute_template_transform() -> None:
     code = """{{args2}}"""
     config = {
         "id": "1",
@@ -92,7 +95,7 @@ def test_execute_template_transform():
         data=TemplateTransformNodeData.model_validate(config["data"]),
         graph_init_params=init_params,
         graph_runtime_state=graph_runtime_state,
-        jinja2_template_renderer=_SimpleJinja2Renderer(),
+        jinja2_template_renderer=cast(Jinja2TemplateRenderer, _SimpleJinja2Renderer()),
     )
 
     # execute node

api/tests/unit_tests/extensions/storage/test_clickzetta_volume_paths.py

@@ -0,0 +1,64 @@
+"""Unit coverage for ClickZetta configuration and path generation."""
+
+import pytest
+
+from extensions.storage.clickzetta_volume.clickzetta_volume_storage import (
+    ClickZettaVolumeConfig,
+    ClickZettaVolumeStorage,
+)
+
+
+def _table_storage() -> ClickZettaVolumeStorage:
+    config = ClickZettaVolumeConfig(
+        username="test_user",
+        password="test_pass",
+        instance="test_instance",
+        service="uat-api.clickzetta.com",
+        workspace="quick_start",
+        vcluster="default_ap",
+        schema_name="dify",
+        volume_type="table",
+        table_prefix="test_dataset_",
+    )
+    storage = ClickZettaVolumeStorage.__new__(ClickZettaVolumeStorage)
+    storage._config = config
+    return storage
+
+
+def test_config_validation() -> None:
+    with pytest.raises(ValueError):
+        ClickZettaVolumeConfig(username="", password="pass", instance="instance")
+
+    with pytest.raises(ValueError):
+        ClickZettaVolumeConfig(username="user", password="pass", instance="instance", volume_type="invalid_type")
+
+    with pytest.raises(ValueError):
+        ClickZettaVolumeConfig(
+            username="user",
+            password="pass",
+            instance="instance",
+            volume_type="external",
+        )
+
+
+def test_volume_path_generation() -> None:
+    storage = _table_storage()
+
+    assert storage._get_volume_path("test.txt", "12345") == "test_dataset_12345/test.txt"
+    assert storage._get_volume_path("12345/test.txt") == "12345/test.txt"
+
+    storage._config.volume_type = "user"
+    assert storage._get_volume_path("test.txt") == "dify_km/test.txt"
+
+
+def test_sql_prefix_generation() -> None:
+    storage = _table_storage()
+
+    assert storage._get_volume_sql_prefix("12345") == "TABLE VOLUME test_dataset_12345"
+
+    storage._config.volume_type = "user"
+    assert storage._get_volume_sql_prefix() == "USER VOLUME"
+
+    storage._config.volume_type = "external"
+    storage._config.volume_name = "my_external_volume"
+    assert storage._get_volume_sql_prefix() == "VOLUME my_external_volume"

api/tests/unit_tests/models/test_workflow_feature_compatibility.py

@@ -1,12 +1,14 @@
 """
-This test file is used to verify the compatibility of Workflow before and after supporting multiple file types.
+This unit test verifies Workflow compatibility before and after supporting multiple file types.
 """
 
 import json
 
+from pydantic import JsonValue
+
 from models import Workflow
 
-OLD_VERSION_WORKFLOW_FEATURES = {
+OLD_VERSION_WORKFLOW_FEATURES: dict[str, JsonValue] = {
     "file_upload": {
         "image": {
             "enabled": True,
@@ -23,7 +25,7 @@ OLD_VERSION_WORKFLOW_FEATURES = {
     "text_to_speech": {"enabled": False, "language": "", "voice": ""},
 }
 
-NEW_VERSION_WORKFLOW_FEATURES = {
+NEW_VERSION_WORKFLOW_FEATURES: dict[str, JsonValue] = {
     "file_upload": {
         "enabled": True,
         "allowed_file_types": ["image"],
@@ -41,7 +43,7 @@ NEW_VERSION_WORKFLOW_FEATURES = {
 }
 
 
-def test_workflow_features():
+def test_workflow_features() -> None:
     workflow = Workflow(
         tenant_id="",
         app_id="",

api/tests/unit_tests/utils/module_import_fixtures/child_class.py

@@ -0,0 +1,7 @@
+from tests.unit_tests.utils.module_import_fixtures.parent_class import ParentClass
+
+
+class ChildClass(ParentClass):
+    def __init__(self, name: str) -> None:
+        super().__init__(name)
+        self.name = name

api/tests/unit_tests/utils/module_import_fixtures/lazy_load_class.py

@@ -0,0 +1,7 @@
+from tests.unit_tests.utils.module_import_fixtures.parent_class import ParentClass
+
+
+class LazyLoadChildClass(ParentClass):
+    def __init__(self, name: str) -> None:
+        super().__init__(name)
+        self.name = name

api/tests/unit_tests/utils/module_import_fixtures/parent_class.py

@@ -0,0 +1,6 @@
+class ParentClass:
+    def __init__(self, name: str) -> None:
+        self.name = name
+
+    def get_name(self) -> str:
+        return self.name

api/tests/unit_tests/utils/test_module_import_helper_behavior.py

@@ -0,0 +1,31 @@
+from pathlib import Path
+
+from core.helper.module_import_helper import import_module_from_source, load_single_subclass_from_source
+from tests.unit_tests.utils.module_import_fixtures.parent_class import ParentClass
+
+FIXTURE_DIR = Path(__file__).parent / "module_import_fixtures"
+
+
+def test_loading_subclass_from_source() -> None:
+    module = load_single_subclass_from_source(
+        module_name="ChildClass", script_path=str(FIXTURE_DIR / "child_class.py"), parent_type=ParentClass
+    )
+    assert module
+    assert module.__name__ == "ChildClass"
+
+
+def test_load_import_module_from_source() -> None:
+    module = import_module_from_source(module_name="ChildClass", py_file_path=str(FIXTURE_DIR / "child_class.py"))
+    assert module
+    assert module.__name__ == "ChildClass"
+
+
+def test_lazy_loading_subclass_from_source() -> None:
+    clz = load_single_subclass_from_source(
+        module_name="LazyLoadChildClass",
+        script_path=str(FIXTURE_DIR / "lazy_load_class.py"),
+        parent_type=ParentClass,
+        use_lazy_loader=True,
+    )
+    instance = clz("dify")
+    assert instance.get_name() == "dify"