From 92b8044cb3cee738a791b3edde4ae3088b70e862 Mon Sep 17 00:00:00 2001
From: CI Bot <ci-bot@xiaoxiajianji.com>
Date: Wed, 22 Jul 2026 14:31:01 +0800
Subject: [PATCH] =?UTF-8?q?test(storage):=20=E8=A1=A5=E5=85=85SharedStorag?=
 =?UTF-8?q?eService=2055=E4=B8=AA=E5=8D=95=E5=85=83=E6=B5=8B=E8=AF=95?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

覆盖 _normalize_storage_key/URL解码、_is_local_generated_url、
get_url、create_direct_upload_post/policy+HMAC签名验证、
get_download_url fallback、未配置OSS错误处理、
bucket操作调用验证、单例模式、endpoint https前缀处理
---
 tests/unit/test_storage_service.py | 577 +++++++++++++++++++++++++++++
 1 file changed, 577 insertions(+)
 create mode 100755 tests/unit/test_storage_service.py

diff --git a/tests/unit/test_storage_service.py b/tests/unit/test_storage_service.py
new file mode 100755
index 000000000..9ce15dac9
--- /dev/null
+++ b/tests/unit/test_storage_service.py
@@ -0,0 +1,577 @@
+"""
+SharedStorageService 单元测试
+
+重点覆盖纯逻辑部分：
+- _normalize_storage_key: URL提取 + URL解码
+- _is_local_generated_url: 本地生成URL判断
+- get_url: 公共URL拼接
+- create_direct_upload_post: policy + HMAC签名
+- get_download_url: bucket=None时的fallback
+- 未配置OSS时的错误处理
+- 单例模式
+"""
+
+import base64
+import hashlib
+import hmac
+import json
+import os
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from packages.shared.storage import (
+    SharedStorageService,
+    get_shared_storage_service,
+    get_storage_service,
+)
+
+
+# ============================================================
+# Fixtures
+# ============================================================
+
+
+def _make_service(
+    bucket_name="test-bucket",
+    endpoint="oss-cn-hangzhou.aliyuncs.com",
+    access_key_id="test-key-id",
+    access_key_secret="test-key-secret",
+    local_url_prefix="/generated-files",
+    with_bucket=True,
+):
+    """创建一个 SharedStorageService 实例，mock 掉 oss2 和 settings。"""
+    mock_settings = MagicMock()
+    mock_settings.oss_bucket_name = bucket_name
+    mock_settings.oss_endpoint = endpoint
+    mock_settings.oss_access_key_id = access_key_id
+    mock_settings.oss_access_key_secret = access_key_secret
+
+    mock_bucket = MagicMock() if with_bucket else None
+
+    with patch("packages.shared.storage.get_shared_settings", return_value=mock_settings), \
+         patch.dict(os.environ, {"GENERATED_FILES_URL_PREFIX": local_url_prefix}, clear=False):
+
+        if with_bucket:
+            with patch("packages.shared.storage.oss2") as mock_oss2:
+                mock_oss2.Auth.return_value = MagicMock()
+                mock_oss2.Bucket.return_value = mock_bucket
+                service = SharedStorageService()
+                service.bucket = mock_bucket
+                return service, mock_bucket, mock_settings
+        else:
+            service = SharedStorageService()
+            service.bucket = None
+            return service, None, mock_settings
+
+
+# ============================================================
+# _normalize_storage_key
+# ============================================================
+
+
+class TestNormalizeStorageKey:
+    """_normalize_storage_key URL 提取与解码"""
+
+    def test_plain_key_returns_as_is(self):
+        service, _, _ = _make_service()
+        result = service._normalize_storage_key("videos/clip.mp4")
+        assert result == "videos/clip.mp4"
+
+    def test_key_with_leading_slash_stripped(self):
+        service, _, _ = _make_service()
+        result = service._normalize_storage_key("/videos/clip.mp4")
+        assert result == "videos/clip.mp4"
+
+    def test_https_url_extracts_path(self):
+        service, _, _ = _make_service()
+        result = service._normalize_storage_key(
+            "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/videos/clip.mp4"
+        )
+        assert result == "videos/clip.mp4"
+
+    def test_http_url_extracts_path(self):
+        service, _, _ = _make_service()
+        result = service._normalize_storage_key(
+            "http://test-bucket.oss-cn-hangzhou.aliyuncs.com/audio/voice.mp3"
+        )
+        assert result == "audio/voice.mp3"
+
+    def test_url_with_query_strips_query(self):
+        service, _, _ = _make_service()
+        result = service._normalize_storage_key(
+            "https://bucket.oss-cn.com/file.mp4?signature=abc&expires=123"
+        )
+        assert result == "file.mp4"
+
+    def test_url_with_leading_slash_in_path(self):
+        service, _, _ = _make_service()
+        result = service._normalize_storage_key(
+            "https://bucket.oss.com//double/slash.jpg"
+        )
+        assert result == "double/slash.jpg"
+
+    def test_url_decodes_percent_encoded_spaces(self):
+        service, _, _ = _make_service()
+        result = service._normalize_storage_key(
+            "https://bucket.oss.com/my%20video.mp4"
+        )
+        assert result == "my video.mp4"
+
+    def test_url_decodes_percent_encoded_chinese(self):
+        service, _, _ = _make_service()
+        result = service._normalize_storage_key(
+            "https://bucket.oss.com/%E4%B8%AD%E6%96%87.mp4"
+        )
+        assert result == "中文.mp4"
+
+    def test_url_with_special_chars_decoded(self):
+        service, _, _ = _make_service()
+        result = service._normalize_storage_key(
+            "https://bucket.oss.com/file%281%29.jpg"
+        )
+        assert result == "file(1).jpg"
+
+    def test_plain_key_with_percent_not_decoded(self):
+        """原始 key 不以 http 开头，不做 URL 解码，直接 lstrip('/')"""
+        service, _, _ = _make_service()
+        result = service._normalize_storage_key("file%20name.mp4")
+        # 不是 URL，直接返回（去掉前导/）
+        assert result == "file%20name.mp4"
+
+    def test_empty_string(self):
+        service, _, _ = _make_service()
+        result = service._normalize_storage_key("")
+        assert result == ""
+
+    def test_root_slash_url(self):
+        service, _, _ = _make_service()
+        result = service._normalize_storage_key("https://bucket.oss.com/")
+        assert result == ""
+
+    def test_nested_path_url(self):
+        service, _, _ = _make_service()
+        result = service._normalize_storage_key(
+            "https://bucket.oss.com/a/b/c/d/file.txt"
+        )
+        assert result == "a/b/c/d/file.txt"
+
+    def test_url_with_port(self):
+        service, _, _ = _make_service()
+        result = service._normalize_storage_key(
+            "https://bucket.oss.com:443/file.txt"
+        )
+        assert result == "file.txt"
+
+
+# ============================================================
+# _is_local_generated_url
+# ============================================================
+
+
+class TestIsLocalGeneratedUrl:
+    """_is_local_generated_url 本地URL判断"""
+
+    def test_local_prefix_returns_true(self):
+        service, _, _ = _make_service()
+        assert service._is_local_generated_url("/generated-files/abc.mp4") is True
+
+    def test_relative_local_returns_true(self):
+        service, _, _ = _make_service()
+        # 没有 scheme，直接用原字符串匹配
+        assert service._is_local_generated_url("/generated-files/out.mp4") is True
+
+    def test_full_url_with_local_path_returns_true(self):
+        service, _, _ = _make_service()
+        assert service._is_local_generated_url(
+            "https://example.com/generated-files/abc.mp4"
+        ) is True
+
+    def test_other_path_returns_false(self):
+        service, _, _ = _make_service()
+        assert service._is_local_generated_url("/videos/abc.mp4") is False
+
+    def test_empty_string_returns_false(self):
+        service, _, _ = _make_service()
+        assert service._is_local_generated_url("") is False
+
+    def test_custom_prefix(self):
+        service, _, _ = _make_service(local_url_prefix="/custom-prefix")
+        assert service._is_local_generated_url("/custom-prefix/file.mp4") is True
+        assert service._is_local_generated_url("/generated-files/file.mp4") is False
+
+
+# ============================================================
+# get_url
+# ============================================================
+
+
+class TestGetUrl:
+    """get_url 公共URL拼接"""
+
+    def test_returns_public_url_plus_key(self):
+        service, _, _ = _make_service()
+        result = service.get_url("videos/test.mp4")
+        assert result == "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/videos/test.mp4"
+
+    def test_empty_key(self):
+        service, _, _ = _make_service()
+        result = service.get_url("")
+        assert result == "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/"
+
+    def test_custom_bucket_and_endpoint(self):
+        service, _, _ = _make_service(
+            bucket_name="my-bucket",
+            endpoint="oss-us-east-1.aliyuncs.com",
+        )
+        result = service.get_url("file.txt")
+        assert result == "https://my-bucket.oss-us-east-1.aliyuncs.com/file.txt"
+
+
+# ============================================================
+# create_direct_upload_post
+# ============================================================
+
+
+class TestCreateDirectUploadPost:
+    """create_direct_upload_post 直传表单生成"""
+
+    def test_returns_dict_with_expected_keys(self):
+        service, _, _ = _make_service()
+        result = service.create_direct_upload_post(
+            storage_key="uploads/test.jpg",
+            content_type="image/jpeg",
+            max_size_bytes=10 * 1024 * 1024,
+            expires_seconds=3600,
+        )
+        assert "url" in result
+        assert "method" in result
+        assert "storage_key" in result
+        assert "expires_at" in result
+        assert "fields" in result
+
+    def test_method_is_post(self):
+        service, _, _ = _make_service()
+        result = service.create_direct_upload_post(
+            "uploads/a.jpg", "image/jpeg", 1024, 3600
+        )
+        assert result["method"] == "POST"
+
+    def test_url_is_public_url(self):
+        service, _, _ = _make_service()
+        result = service.create_direct_upload_post(
+            "uploads/a.jpg", "image/jpeg", 1024, 3600
+        )
+        assert result["url"] == "https://test-bucket.oss-cn-hangzhou.aliyuncs.com"
+
+    def test_storage_key_normalized(self):
+        service, _, _ = _make_service()
+        result = service.create_direct_upload_post(
+            "/uploads/test.jpg", "image/jpeg", 1024, 3600
+        )
+        assert result["storage_key"] == "uploads/test.jpg"
+        assert result["fields"]["key"] == "uploads/test.jpg"
+
+    def test_fields_contain_required_keys(self):
+        service, _, _ = _make_service()
+        result = service.create_direct_upload_post(
+            "uploads/a.jpg", "image/jpeg", 1024, 3600
+        )
+        fields = result["fields"]
+        assert fields["key"] == "uploads/a.jpg"
+        assert fields["OSSAccessKeyId"] == "test-key-id"
+        assert fields["success_action_status"] == "201"
+        assert fields["Content-Type"] == "image/jpeg"
+        assert "policy" in fields
+        assert "Signature" in fields
+
+    def test_policy_signature_is_valid_hmac_sha1(self):
+        """验证 HMAC-SHA1 签名是否正确"""
+        secret = "my-secret-key-123"
+        service, _, _ = _make_service(access_key_secret=secret)
+        result = service.create_direct_upload_post(
+            "uploads/a.jpg", "image/jpeg", 1024, 3600
+        )
+        policy = result["fields"]["policy"]
+        signature = result["fields"]["Signature"]
+
+        # 手动计算签名验证
+        expected = base64.b64encode(
+            hmac.new(secret.encode("utf-8"), policy.encode("utf-8"), hashlib.sha1).digest()
+        ).decode("ascii")
+        assert signature == expected
+
+    def test_policy_contains_bucket_and_key(self):
+        service, _, _ = _make_service(bucket_name="my-bucket")
+        result = service.create_direct_upload_post(
+            "uploads/photo.png", "image/png", 2048, 1800
+        )
+        policy = json.loads(base64.b64decode(result["fields"]["policy"]))
+        conditions = policy["conditions"]
+
+        assert {"bucket": "my-bucket"} in conditions
+        assert {"key": "uploads/photo.png"} in conditions
+
+    def test_policy_contains_content_length_range(self):
+        service, _, _ = _make_service()
+        result = service.create_direct_upload_post(
+            "uploads/a.jpg", "image/jpeg", 5242880, 3600
+        )
+        policy = json.loads(base64.b64decode(result["fields"]["policy"]))
+        conditions = policy["conditions"]
+
+        size_condition = [c for c in conditions if isinstance(c, list) and c[0] == "content-length-range"]
+        assert len(size_condition) == 1
+        assert size_condition[0][1] == 1
+        assert size_condition[0][2] == 5242880
+
+    def test_policy_content_type_starts_with(self):
+        service, _, _ = _make_service()
+        result = service.create_direct_upload_post(
+            "uploads/a.jpg", "image/jpeg", 1024, 3600
+        )
+        policy = json.loads(base64.b64decode(result["fields"]["policy"]))
+        conditions = policy["conditions"]
+
+        ct_condition = [c for c in conditions if isinstance(c, list) and c[0] == "starts-with"]
+        assert len(ct_condition) == 1
+        assert ct_condition[0][1] == "$Content-Type"
+        assert ct_condition[0][2] == "image/"
+
+    def test_policy_has_expiration(self):
+        service, _, _ = _make_service()
+        result = service.create_direct_upload_post(
+            "uploads/a.jpg", "image/jpeg", 1024, 3600
+        )
+        policy = json.loads(base64.b64decode(result["fields"]["policy"]))
+        assert "expiration" in policy
+        # ISO 8601 格式
+        assert policy["expiration"].endswith("Z")
+
+    def test_non_uploads_key_raises_value_error(self):
+        service, _, _ = _make_service()
+        with pytest.raises(ValueError, match="uploads/"):
+            service.create_direct_upload_post(
+                "videos/a.mp4", "video/mp4", 1024, 3600
+            )
+
+    def test_no_credentials_raises_runtime_error(self):
+        service, _, _ = _make_service(access_key_id="", access_key_secret="", with_bucket=False)
+        with pytest.raises(RuntimeError, match="not configured"):
+            service.create_direct_upload_post(
+                "uploads/a.jpg", "image/jpeg", 1024, 3600
+            )
+
+    def test_url_normalized_key_in_uploads(self):
+        service, _, _ = _make_service()
+        # URL 形式的 key 被 normalize 后如果在 uploads/ 下应该可以
+        result = service.create_direct_upload_post(
+            "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/uploads/from_url.jpg",
+            "image/jpeg",
+            1024,
+            3600,
+        )
+        assert result["storage_key"] == "uploads/from_url.jpg"
+
+
+# ============================================================
+# get_download_url (bucket=None 时的 fallback)
+# ============================================================
+
+
+class TestGetDownloadUrlFallback:
+    """get_download_url 在 bucket 未配置时的 fallback 逻辑"""
+
+    def test_no_bucket_local_url_returns_as_is(self):
+        service, _, _ = _make_service(with_bucket=False)
+        result = service.get_download_url("/generated-files/test.mp4")
+        assert result == "/generated-files/test.mp4"
+
+    def test_no_bucket_regular_key_returns_public_url(self):
+        service, _, _ = _make_service(with_bucket=False)
+        result = service.get_download_url("videos/test.mp4")
+        assert result == "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/videos/test.mp4"
+
+    def test_no_bucket_url_input_normalized(self):
+        service, _, _ = _make_service(with_bucket=False)
+        result = service.get_download_url(
+            "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/videos/test.mp4"
+        )
+        assert result == "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/videos/test.mp4"
+
+    def test_with_bucket_calls_sign_url(self):
+        service, mock_bucket, _ = _make_service(with_bucket=True)
+        mock_bucket.sign_url.return_value = "https://signed-url.com/file?sig=abc"
+
+        result = service.get_download_url("videos/test.mp4", expires_seconds=7200)
+
+        mock_bucket.sign_url.assert_called_once_with("GET", "videos/test.mp4", 7200)
+        assert result == "https://signed-url.com/file?sig=abc"
+
+    def test_sign_url_exception_falls_back_to_public_url(self):
+        service, mock_bucket, _ = _make_service(with_bucket=True)
+        mock_bucket.sign_url.side_effect = Exception("sign error")
+
+        result = service.get_download_url("videos/test.mp4")
+
+        assert result == "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/videos/test.mp4"
+
+
+# ============================================================
+# 未配置 OSS 时的错误处理
+# ============================================================
+
+
+class TestNoBucketErrorHandling:
+    """bucket=None 时的错误处理"""
+
+    def test_upload_file_raises(self):
+        service, _, _ = _make_service(with_bucket=False)
+        with pytest.raises(RuntimeError, match="not configured"):
+            service.upload_file("/tmp/test.txt", "uploads/test.txt")
+
+    def test_download_file_raises(self):
+        service, _, _ = _make_service(with_bucket=False)
+        with pytest.raises(RuntimeError, match="not configured"):
+            service.download_file("uploads/test.txt", "/tmp/test.txt")
+
+    def test_delete_file_silent_noop(self):
+        service, _, _ = _make_service(with_bucket=False)
+        # 不抛异常
+        result = service.delete_file("uploads/test.txt")
+        assert result is None
+
+    def test_file_exists_returns_false(self):
+        service, _, _ = _make_service(with_bucket=False)
+        assert service.file_exists("uploads/test.txt") is False
+
+
+# ============================================================
+# upload_file / delete_file / file_exists 正常路径
+# ============================================================
+
+
+class TestBucketOperations:
+    """有 bucket 时的操作调用验证"""
+
+    def test_upload_file_with_path_string(self):
+        service, mock_bucket, _ = _make_service()
+        result = service.upload_file("/tmp/file.txt", "uploads/file.txt", "text/plain")
+
+        mock_bucket.put_object_from_file.assert_called_once()
+        args = mock_bucket.put_object_from_file.call_args
+        assert args[0][0] == "uploads/file.txt"
+        assert args[0][1] == "/tmp/file.txt"
+        assert result.startswith("https://test-bucket.")
+
+    def test_upload_file_with_file_object(self):
+        service, mock_bucket, _ = _make_service()
+        mock_file = MagicMock()
+        result = service.upload_file(mock_file, "uploads/file.bin", "application/octet-stream")
+
+        mock_file.seek.assert_called_once_with(0)
+        mock_bucket.put_object.assert_called_once()
+        assert result.startswith("https://test-bucket.")
+
+    def test_delete_file_calls_bucket(self):
+        service, mock_bucket, _ = _make_service()
+        service.delete_file("uploads/test.txt")
+        mock_bucket.delete_object.assert_called_once_with("uploads/test.txt")
+
+    def test_delete_file_exception_logged_not_raised(self):
+        service, mock_bucket, _ = _make_service()
+        mock_bucket.delete_object.side_effect = Exception("delete error")
+        # 不抛异常
+        service.delete_file("uploads/test.txt")
+
+    def test_file_exists_delegates_to_bucket(self):
+        service, mock_bucket, _ = _make_service()
+        mock_bucket.object_exists.return_value = True
+        assert service.file_exists("some/key") is True
+        mock_bucket.object_exists.assert_called_once_with("some/key")
+
+    def test_file_exists_false(self):
+        service, mock_bucket, _ = _make_service()
+        mock_bucket.object_exists.return_value = False
+        assert service.file_exists("some/key") is False
+
+
+# ============================================================
+# 单例 & 兼容别名
+# ============================================================
+
+
+class TestSingleton:
+    """get_shared_storage_service 单例模式"""
+
+    def test_get_storage_service_is_alias(self):
+        # 两个函数返回同一个实例
+        with patch("packages.shared.storage._storage_service", None):
+            with patch("packages.shared.storage.SharedStorageService") as mock_cls:
+                mock_instance = MagicMock()
+                mock_cls.return_value = mock_instance
+
+                svc1 = get_shared_storage_service()
+                svc2 = get_storage_service()
+
+                assert svc1 is svc2
+                # 因为是同一个单例，类只实例化一次
+                assert mock_cls.call_count == 1
+
+
+# ============================================================
+# __init__ endpoint 处理
+# ============================================================
+
+
+class TestInitEndpointHandling:
+    """初始化时 endpoint https 前缀处理"""
+
+    def test_endpoint_without_https_gets_prefix(self):
+        mock_settings = MagicMock()
+        mock_settings.oss_bucket_name = "test-bucket"
+        mock_settings.oss_endpoint = "oss-cn-hangzhou.aliyuncs.com"
+        mock_settings.oss_access_key_id = "key-id"
+        mock_settings.oss_access_key_secret = "key-secret"
+
+        with patch("packages.shared.storage.get_shared_settings", return_value=mock_settings), \
+             patch("packages.shared.storage.oss2") as mock_oss2:
+            mock_oss2.Bucket.return_value = MagicMock()
+
+            service = SharedStorageService()
+
+            # 验证 Bucket 构造时 endpoint 带了 https://
+            call_args = mock_oss2.Bucket.call_args
+            assert call_args[0][1] == "https://oss-cn-hangzhou.aliyuncs.com"
+
+    def test_endpoint_with_https_keeps_as_is(self):
+        mock_settings = MagicMock()
+        mock_settings.oss_bucket_name = "test-bucket"
+        mock_settings.oss_endpoint = "https://oss-cn-hangzhou.aliyuncs.com"
+        mock_settings.oss_access_key_id = "key-id"
+        mock_settings.oss_access_key_secret = "key-secret"
+
+        with patch("packages.shared.storage.get_shared_settings", return_value=mock_settings), \
+             patch("packages.shared.storage.oss2") as mock_oss2:
+            mock_oss2.Bucket.return_value = MagicMock()
+
+            service = SharedStorageService()
+
+            call_args = mock_oss2.Bucket.call_args
+            assert call_args[0][1] == "https://oss-cn-hangzhou.aliyuncs.com"
+
+    def test_endpoint_with_http_keeps_as_is(self):
+        mock_settings = MagicMock()
+        mock_settings.oss_bucket_name = "test-bucket"
+        mock_settings.oss_endpoint = "http://oss-cn-hangzhou.aliyuncs.com"
+        mock_settings.oss_access_key_id = "key-id"
+        mock_settings.oss_access_key_secret = "key-secret"
+
+        with patch("packages.shared.storage.get_shared_settings", return_value=mock_settings), \
+             patch("packages.shared.storage.oss2") as mock_oss2:
+            mock_oss2.Bucket.return_value = MagicMock()
+
+            service = SharedStorageService()
+
+            call_args = mock_oss2.Bucket.call_args
+            assert call_args[0][1] == "http://oss-cn-hangzhou.aliyuncs.com"