test: add Baidu OBS storage unit tests (#34330)

api/tests/unit_tests/oss/__mock/baidu_obs.py

@@ -0,0 +1,70 @@
+import base64
+import hashlib
+import os
+from io import BytesIO
+from types import SimpleNamespace
+
+import pytest
+from _pytest.monkeypatch import MonkeyPatch
+from baidubce.services.bos.bos_client import BosClient
+
+from tests.unit_tests.oss.__mock.base import (
+    get_example_bucket,
+    get_example_data,
+    get_example_filename,
+    get_example_filepath,
+)
+
+
+class MockBaiduObsClass:
+    def __init__(self, config=None):
+        self.bucket_name = get_example_bucket()
+        self.key = get_example_filename()
+        self.content = get_example_data()
+        self.filepath = get_example_filepath()
+
+    def put_object(self, bucket_name, key, data, content_length=None, content_md5=None, **kwargs):
+        assert bucket_name == self.bucket_name
+        assert key == self.key
+        assert data == self.content
+        assert content_length == len(self.content)
+        expected_md5 = base64.standard_b64encode(hashlib.md5(self.content).digest())
+        assert content_md5 == expected_md5
+
+    def get_object(self, bucket_name, key, **kwargs):
+        assert bucket_name == self.bucket_name
+        assert key == self.key
+        return SimpleNamespace(data=BytesIO(self.content))
+
+    def get_object_to_file(self, bucket_name, key, file_name, **kwargs):
+        assert bucket_name == self.bucket_name
+        assert key == self.key
+        assert file_name == self.filepath
+
+    def get_object_meta_data(self, bucket_name, key, **kwargs):
+        assert bucket_name == self.bucket_name
+        assert key == self.key
+        return SimpleNamespace(status=200)
+
+    def delete_object(self, bucket_name, key, **kwargs):
+        assert bucket_name == self.bucket_name
+        assert key == self.key
+
+
+MOCK = os.getenv("MOCK_SWITCH", "false").lower() == "true"
+
+
+@pytest.fixture
+def setup_baidu_obs_mock(monkeypatch: MonkeyPatch):
+    if MOCK:
+        monkeypatch.setattr(BosClient, "__init__", MockBaiduObsClass.__init__)
+        monkeypatch.setattr(BosClient, "put_object", MockBaiduObsClass.put_object)
+        monkeypatch.setattr(BosClient, "get_object", MockBaiduObsClass.get_object)
+        monkeypatch.setattr(BosClient, "get_object_to_file", MockBaiduObsClass.get_object_to_file)
+        monkeypatch.setattr(BosClient, "get_object_meta_data", MockBaiduObsClass.get_object_meta_data)
+        monkeypatch.setattr(BosClient, "delete_object", MockBaiduObsClass.delete_object)
+
+    yield
+
+    if MOCK:
+        monkeypatch.undo()

api/tests/unit_tests/oss/baidu_obs/__init__.py

@@ -0,0 +1 @@
+

api/tests/unit_tests/oss/baidu_obs/test_baidu_obs.py

@@ -0,0 +1,59 @@
+from unittest.mock import MagicMock, patch
+
+import pytest
+from baidubce.auth.bce_credentials import BceCredentials
+from baidubce.bce_client_configuration import BceClientConfiguration
+
+from extensions.storage.baidu_obs_storage import BaiduObsStorage
+from tests.unit_tests.oss.__mock.baidu_obs import setup_baidu_obs_mock
+from tests.unit_tests.oss.__mock.base import (
+    BaseStorageTest,
+    get_example_bucket,
+)
+
+
+class TestBaiduObs(BaseStorageTest):
+    @pytest.fixture(autouse=True)
+    def setup_method(self, setup_baidu_obs_mock):
+        """Executed before each test method."""
+        with (
+            patch.object(BceCredentials, "__init__", return_value=None),
+            patch.object(BceClientConfiguration, "__init__", return_value=None),
+        ):
+            self.storage = BaiduObsStorage()
+        self.storage.bucket_name = get_example_bucket()
+
+
+class TestBaiduObsConfiguration:
+    def test_init_with_config(self):
+        mock_dify_config = MagicMock()
+        mock_dify_config.BAIDU_OBS_BUCKET_NAME = "test-bucket"
+        mock_dify_config.BAIDU_OBS_ACCESS_KEY = "test-access-key"
+        mock_dify_config.BAIDU_OBS_SECRET_KEY = "test-secret-key"
+        mock_dify_config.BAIDU_OBS_ENDPOINT = "https://bj.bcebos.com"
+
+        mock_credentials = MagicMock(name="credentials")
+        mock_config = MagicMock(name="config")
+        mock_client = MagicMock(name="client")
+
+        with (
+            patch("extensions.storage.baidu_obs_storage.dify_config", mock_dify_config),
+            patch("extensions.storage.baidu_obs_storage.BceCredentials", return_value=mock_credentials) as credentials,
+            patch(
+                "extensions.storage.baidu_obs_storage.BceClientConfiguration", return_value=mock_config
+            ) as configuration,
+            patch("extensions.storage.baidu_obs_storage.BosClient", return_value=mock_client) as client_cls,
+        ):
+            storage = BaiduObsStorage()
+
+        assert storage.bucket_name == "test-bucket"
+        assert storage.client == mock_client
+        credentials.assert_called_once_with(
+            access_key_id="test-access-key",
+            secret_access_key="test-secret-key",
+        )
+        configuration.assert_called_once_with(
+            credentials=mock_credentials,
+            endpoint="https://bj.bcebos.com",
+        )
+        client_cls.assert_called_once_with(config=mock_config)