xiaoxia
34ea6a866b
CI/CD Pipeline / PR Build API Image (push) Has been skipped
CI/CD Pipeline / PR Build Web Image (push) Has been skipped
CI/CD Pipeline / PR Build Worker Image (push) Has been skipped
CI/CD Pipeline / Build Production API Image (push) Has been skipped
CI/CD Pipeline / Build Production Web Image (push) Has been skipped
CI/CD Pipeline / Build Production Worker Image (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Build Staging Web Image (push) Successful in 25s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 1m15s
CI/CD Pipeline / Build Staging API Image (push) Successful in 4m56s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 41s
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 49s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 4m37s
CI/CD Pipeline / Staging E2E Tests (push) Failing after 4m45s
CI/CD Pipeline / Check if frontend-only change (push) Has been skipped
CI/CD Pipeline / Validate - Type Check (mypy) (push) Successful in 15s
CI/CD Pipeline / Validate - Migration (alembic) (push) Successful in 7s
CI/CD Pipeline / Frontend Lint (push) Successful in 24s
CI/CD Pipeline / Validate - Code Quality (push) Failing after 58s
CI/CD Pipeline / Frontend Unit Tests (push) Failing after 6s
CI/CD Pipeline / Unit Tests (push) Successful in 2m16s
CI/CD Pipeline / Integration Tests (push) Failing after 2m17s
Co-authored-by: xiaoxia <dev@xiaoxiajianji.com> Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>
540 lines
21 KiB
Python
Executable File
540 lines
21 KiB
Python
Executable File
"""
|
|
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"
|