test: move pure integration cases to unit tests

api/tests/integration_tests/storage/test_clickzetta_volume.py

@@ -103,66 +103,5 @@ 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/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

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

@@ -1,3 +1,5 @@
+"""Unit coverage for datasource node event streaming."""
+
 from pytest_mock import MockerFixture
 
 from core.app.entities.app_invoke_entities import DIFY_RUN_CONTEXT_KEY

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

@@ -1,3 +1,5 @@
+"""Unit coverage for in-process template-transform execution."""
+
 import time
 import uuid
 

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,5 +1,5 @@
 """
-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

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

@@ -1,4 +1,4 @@
-from tests.integration_tests.utils.parent_class import ParentClass
+from tests.unit_tests.utils.module_import_fixtures.parent_class import ParentClass
 
 
 class ChildClass(ParentClass):

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

@@ -1,4 +1,4 @@
-from tests.integration_tests.utils.parent_class import ParentClass
+from tests.unit_tests.utils.module_import_fixtures.parent_class import ParentClass
 
 
 class LazyLoadChildClass(ParentClass):

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

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

api/tests/unit_tests/utils/test_module_import_helper_behavior.py

@@ -1,32 +1,29 @@
-import os
+from pathlib import Path
 
 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
+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():
-    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
+        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():
-    current_path = os.getcwd()
-    module = import_module_from_source(
-        module_name="ChildClass", py_file_path=os.path.join(current_path, "child_class.py")
-    )
+    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():
-    current_path = os.getcwd()
     clz = load_single_subclass_from_source(
         module_name="LazyLoadChildClass",
-        script_path=os.path.join(current_path, "lazy_load_class.py"),
+        script_path=str(FIXTURE_DIR / "lazy_load_class.py"),
         parent_type=ParentClass,
         use_lazy_loader=True,
     )