xiaoxia/xiaoxia-saas
1
0
Fork 0
Code Issues 11 Pull Requests 3 Actions 4 Packages Projects Releases 7 Wiki Activity

Phase 4 Backend P0 cleanup #276

Merged
xiaoxia merged 1 commits from cleanup/phase4-backend-p0 into develop 2026-07-13 18:04:49 +08:00
Conversation 0 Commits 1 Files Changed 9
+124 -347
9 changed files with 124 additions and 347 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.production.example
-1
View File
@@ -5,7 +5,6 @@ APP_ENV=production
ENVIRONMENT=production
DEBUG=false
USE_IN_MEMORY_DB=false
LOG_LEVEL=WARNING
# ==================== 数据库（必须修改）====================
DATABASE_URL=postgresql://prod_user:CHANGE_THIS_PASSWORD@db-prod:5432/xiaoxia_prod
apps/api/app/config.py
-8
View File
@@ -19,7 +19,6 @@ class Settings(BaseSettings):
# Container bind address; external expose is controlled by Docker/Nginx.
API_HOST: str = "0.0.0.0" # nosec: B104
API_PORT: int = 8000
API_PREFIX: str = "/api/v1"
DATABASE_URL: str = "postgresql+psycopg://postgres:postgres@localhost:5432/xiaoxia_saas"
DATABASE_POOL_SIZE: int = 20
@@ -30,16 +29,10 @@ class Settings(BaseSettings):
AUTO_CREATE_SCHEMA: bool = False
REDIS_URL: str = "redis://localhost:6379/0"
REDIS_MAX_CONNECTION: int = 50
ENABLE_REDIS_SESSIONS: bool = False
# JWT secret key - MUST be set via environment variable, no default allowed
JWT_SECRET_KEY: Optional[str] = None
# 旧的 JWT secret key（用于密钥轮换期间验证旧 token）
# 在密钥轮换时，先设置新密钥，旧密钥保留在此处直到所有旧 token 过期
JWT_SECRET_KEY_OLD: Optional[str] = None
# 密钥轮换天数（到达此天数后建议更换密钥）
SECRET_ROTATION_DAYS: int = 90
# JWT 算法与过期时间（与 .env.example 对齐）
JWT_ALGORITHM: str = "HS256"
@@ -111,7 +104,6 @@ class Settings(BaseSettings):
)
OSS_DIRECT_UPLOAD_EXPIRE_SECONDS: int = 900
LOG_LEVEL: str = "INFO"
CORS_ORIGINS_RAW: str = "http://localhost:3000,http://localhost:5173,http://localhost:8000"
# 渲染引擎选择：legacy=旧VideoComposeService，unified=新UnifiedRenderService
apps/api/app/core/storage.py
+11 -280
View File
@@ -1,283 +1,14 @@
"""阿里云 OSS 存储服务"""
"""Backward-compatible re-export from shared storage.
import base64
import datetime as dt
import hashlib
import hmac
import json
import logging
import os
from urllib.parse import urlparse
All storage logic now lives in ``packages.shared.storage``.
This module keeps old import paths working so existing code
does not need to change.
"""
try:
import oss2
except ImportError: # pragma: no cover - exercised in minimal local/test environments
oss2 = None
from app.config import get_settings
from packages.shared.storage import (
SharedStorageService as OSSStorageService,
get_shared_storage_service,
get_storage_service,
)
logger = logging.getLogger(__name__)
class OSSStorageService:
"""阿里云 OSS 存储服务"""
def __init__(self):
settings = get_settings()
self.bucket_name = settings.OSS_BUCKET_NAME
self.public_url = f"https://{settings.OSS_BUCKET_NAME}.{settings.OSS_ENDPOINT}"
self.local_url_prefix = os.getenv("GENERATED_FILES_URL_PREFIX", "/generated-files")
self.bucket = None
has_key_id = bool(settings.OSS_ACCESS_KEY_ID)
has_key_secret = bool(settings.OSS_ACCESS_KEY_SECRET)
if has_key_id and has_key_secret:
if oss2 is not None:
try:
# P0-2 修复：oss2.Bucket 的 endpoint 必须带 https:// 前缀，
# 否则 sign_url 默认生成 HTTP URL。
bucket_endpoint = settings.OSS_ENDPOINT
if not bucket_endpoint.startswith(("http://", "https://")):
bucket_endpoint = f"https://{bucket_endpoint}"
auth = oss2.Auth(
settings.OSS_ACCESS_KEY_ID,
settings.OSS_ACCESS_KEY_SECRET,
)
self.bucket = oss2.Bucket(
auth,
bucket_endpoint,
settings.OSS_BUCKET_NAME,
)
logger.info(
"OSS initialized: endpoint=%s bucket=%s",
settings.OSS_ENDPOINT,
settings.OSS_BUCKET_NAME,
)
except Exception as error:
logger.error("Failed to initialize OSS bucket client: %s", error)
else:
logger.error("oss2 SDK is not installed — OSS operations will fail")
else:
missing = []
if not has_key_id:
missing.append("OSS_ACCESS_KEY_ID")
if not has_key_secret:
missing.append("OSS_ACCESS_KEY_SECRET")
logger.error("OSS credentials not configured — missing: %s", ", ".join(missing))
self.access_key_id = settings.OSS_ACCESS_KEY_ID
self.access_key_secret = settings.OSS_ACCESS_KEY_SECRET
self.endpoint = settings.OSS_ENDPOINT
def diagnose(self) -> None:
"""启动诊断：输出 OSS 配置状态，帮助排查预签名 URL 问题。"""
key_id_display = (
f"{self.access_key_id[:4]}...{self.access_key_id[-4:]}" if len(self.access_key_id) > 8 else "(empty)"
)
logger.info(
"[OSS诊断] endpoint=%s bucket_name=%s access_key_id=%s",
self.endpoint,
self.bucket_name,
key_id_display,
)
if self.bucket is None:
logger.error(
"[OSS诊断] ❌ bucket=None — 预签名URL不可用！"
"原因: OSS_ACCESS_KEY_ID/OSS_ACCESS_KEY_SECRET 未配置或 oss2 未安装。"
"请检查服务器 .env 文件（如 /var/lib/xiaoxia-saas-staging/.env）"
)
else:
logger.info("[OSS诊断] ✅ bucket 已配置，预签名URL可用")
def _is_local_generated_url(self, storage_key_or_url: str) -> bool:
parsed = urlparse(storage_key_or_url)
path = parsed.path if parsed.scheme else storage_key_or_url
return path.startswith(f"{self.local_url_prefix}/")
def create_direct_upload_post(
self,
storage_key: str,
content_type: str,
max_size_bytes: int,
expires_seconds: int,
) -> dict[str, object]:
"""创建浏览器直传 OSS 的 PostObject 表单。"""
if not self.access_key_id or not self.access_key_secret:
raise RuntimeError("OSS storage is not configured")
normalized_key = self._normalize_storage_key(storage_key)
if not normalized_key.startswith("uploads/"):
raise ValueError("direct upload key must be under uploads/")
expiration = (dt.datetime.now(dt.timezone.utc) + dt.timedelta(seconds=expires_seconds)).strftime(
"%Y-%m-%dT%H:%M:%S.000Z"
)
policy = {
"expiration": expiration,
"conditions": [
{"bucket": self.bucket_name},
{"key": normalized_key},
["content-length-range", 1, max_size_bytes],
["starts-with", "$Content-Type", content_type.split("/", 1)[0] + "/" if "/" in content_type else ""],
],
}
encoded_policy = base64.b64encode(json.dumps(policy, separators=(",", ":")).encode("utf-8")).decode("ascii")
signature = base64.b64encode(
hmac.new(self.access_key_secret.encode("utf-8"), encoded_policy.encode("utf-8"), hashlib.sha1).digest()
).decode("ascii")
return {
"url": self.public_url,
"method": "POST",
"storage_key": normalized_key,
"expires_at": expiration,
"fields": {
"key": normalized_key,
"OSSAccessKeyId": self.access_key_id,
"policy": encoded_policy,
"Signature": signature,
"success_action_status": "201",
"Content-Type": content_type,
},
}
def upload_file(
self,
file_or_path,
storage_key: str,
content_type: str = "application/octet-stream",
) -> str:
"""
上传文件到 OSS
Args:
file_or_path: 文件对象或本地文件路径
storage_key: 存储键（文件路径）
content_type: 内容类型
Returns:
文件公网 URL
"""
if self.bucket is None:
raise RuntimeError("OSS storage is not configured")
try:
# 如果是字符串路径，从本地文件上传
if isinstance(file_or_path, str):
self.bucket.put_object_from_file(storage_key, file_or_path, headers={"Content-Type": content_type})
else:
# 文件对象
file_or_path.seek(0)
self.bucket.put_object(storage_key, file_or_path, headers={"Content-Type": content_type})
return f"{self.public_url}/{storage_key}"
except Exception as e:
raise Exception(f"Failed to upload file to OSS: {e}")
def get_url(self, storage_key: str) -> str:
"""获取文件公网 URL"""
return f"{self.public_url}/{storage_key}"
def get_download_url(self, storage_key_or_url: str, expires_seconds: int = 3600) -> str:
"""
获取文件下载签名 URL（用于私有文件）
Args:
storage_key_or_url: 存储键或完整 URL
expires_seconds: 过期时间（秒）
Returns:
签名 URL
"""
if self.bucket is None:
if self._is_local_generated_url(storage_key_or_url):
return storage_key_or_url
logger.warning(
"get_download_url: OSS bucket not configured, returning raw URL. " "storage_key_or_url=%s",
storage_key_or_url[:200],
)
return self.get_url(self._normalize_storage_key(storage_key_or_url))
storage_key = self._normalize_storage_key(storage_key_or_url)
try:
signed = self.bucket.sign_url("GET", storage_key, expires_seconds)
logger.info(
"get_download_url: signed URL generated. storage_key=%s url_prefix=%s",
storage_key[:80],
signed[:60],
)
return signed
except Exception:
logger.exception(
"get_download_url: sign_url failed, falling back to raw URL. " "storage_key=%s",
storage_key[:200],
)
return self.get_url(storage_key)
def _normalize_storage_key(self, storage_key_or_url: str) -> str:
"""从 URL 中提取存储键"""
if storage_key_or_url.startswith("http://") or storage_key_or_url.startswith("https://"):
parsed = urlparse(storage_key_or_url)
# 移除开头的 /
return parsed.path.lstrip("/")
return storage_key_or_url.lstrip("/")
def download_file(self, storage_key: str, local_path: str):
"""
从 OSS 下载文件到本地
Args:
storage_key: 存储键
local_path: 本地文件路径
"""
if self.bucket is None:
raise RuntimeError("OSS storage is not configured")
try:
os.makedirs(os.path.dirname(local_path), exist_ok=True)
self.bucket.get_object_to_file(storage_key, local_path)
except Exception as e:
raise Exception(f"Failed to download file from OSS: {e}")
def delete_file(self, storage_key: str):
"""
删除 OSS 文件
Args:
storage_key: 存储键
"""
if self.bucket is None:
return
try:
self.bucket.delete_object(storage_key)
except Exception as error:
logger.warning(
"Failed to delete file from OSS",
extra={"storage_key": storage_key, "error": str(error)},
)
def file_exists(self, storage_key: str) -> bool:
"""
检查文件是否存在
Args:
storage_key: 存储键
Returns:
是否存在
"""
if self.bucket is None:
return False
return self.bucket.object_exists(storage_key)
_storage_service = None
def get_storage_service() -> OSSStorageService:
"""获取存储服务实例（全局单例）"""
global _storage_service
if _storage_service is None:
_storage_service = OSSStorageService()
_storage_service.diagnose()
return _storage_service
__all__ = ["OSSStorageService", "get_storage_service", "get_shared_storage_service"]
packages/shared/storage.py
+72 -11
View File
@@ -30,21 +30,67 @@ class SharedStorageService:
self.local_url_prefix = os.getenv("GENERATED_FILES_URL_PREFIX", "/generated-files")
self.bucket = None
if settings.oss_access_key_id and settings.oss_access_key_secret:
has_key_id = bool(settings.oss_access_key_id)
has_key_secret = bool(settings.oss_access_key_secret)
if has_key_id and has_key_secret:
if oss2 is not None:
auth = oss2.Auth(
settings.oss_access_key_id,
settings.oss_access_key_secret,
)
self.bucket = oss2.Bucket(
auth,
settings.oss_endpoint,
settings.oss_bucket_name,
)
try:
# P0-2 修复：oss2.Bucket 的 endpoint 必须带 https:// 前缀，
# 否则 sign_url 默认生成 HTTP URL。
bucket_endpoint = settings.oss_endpoint
if not bucket_endpoint.startswith(("http://", "https://")):
bucket_endpoint = f"https://{bucket_endpoint}"
auth = oss2.Auth(
settings.oss_access_key_id,
settings.oss_access_key_secret,
)
self.bucket = oss2.Bucket(
auth,
bucket_endpoint,
settings.oss_bucket_name,
)
logger.info(
"OSS initialized: endpoint=%s bucket=%s",
settings.oss_endpoint,
settings.oss_bucket_name,
)
except Exception as error:
logger.error("Failed to initialize OSS bucket client: %s", error)
else:
logger.error("oss2 SDK is not installed — OSS operations will fail")
else:
missing = []
if not has_key_id:
missing.append("OSS_ACCESS_KEY_ID")
if not has_key_secret:
missing.append("OSS_ACCESS_KEY_SECRET")
logger.error("OSS credentials not configured — missing: %s", ", ".join(missing))
self.access_key_id = settings.oss_access_key_id
self.access_key_secret = settings.oss_access_key_secret
self.endpoint = settings.oss_endpoint
def diagnose(self) -> None:
"""启动诊断：输出 OSS 配置状态，帮助排查预签名 URL 问题。"""
key_id_display = (
f"{self.access_key_id[:4]}...{self.access_key_id[-4:]}" if len(self.access_key_id) > 8 else "(empty)"
)
logger.info(
"[OSS诊断] endpoint=%s bucket_name=%s access_key_id=%s",
self.endpoint,
self.bucket_name,
key_id_display,
)
if self.bucket is None:
logger.error(
"[OSS诊断] ❌ bucket=None — 预签名URL不可用！"
"原因: OSS_ACCESS_KEY_ID/OSS_ACCESS_KEY_SECRET 未配置或 oss2 未安装。"
"请检查服务器 .env 文件（如 /var/lib/xiaoxia-saas-staging/.env）"
)
else:
logger.info("[OSS诊断] ✅ bucket 已配置，预签名URL可用")
def _is_local_generated_url(self, storage_key_or_url: str) -> bool:
parsed = urlparse(storage_key_or_url)
path = parsed.path if parsed.scheme else storage_key_or_url
@@ -90,12 +136,26 @@ class SharedStorageService:
if self.bucket is None:
if self._is_local_generated_url(storage_key_or_url):
return storage_key_or_url
logger.warning(
"get_download_url: OSS bucket not configured, returning raw URL. storage_key_or_url=%s",
storage_key_or_url[:200],
)
return self.get_url(self._normalize_storage_key(storage_key_or_url))
storage_key = self._normalize_storage_key(storage_key_or_url)
try:
return self.bucket.sign_url("GET", storage_key, expires_seconds)
signed = self.bucket.sign_url("GET", storage_key, expires_seconds)
logger.info(
"get_download_url: signed URL generated. storage_key=%s url_prefix=%s",
storage_key[:80],
signed[:60],
)
return signed
except Exception:
logger.exception(
"get_download_url: sign_url failed, falling back to raw URL. storage_key=%s",
storage_key[:200],
)
return self.get_url(storage_key)
def _normalize_storage_key(self, storage_key_or_url: str) -> str:
@@ -175,6 +235,7 @@ def get_shared_storage_service() -> SharedStorageService:
global _storage_service
if _storage_service is None:
_storage_service = SharedStorageService()
_storage_service.diagnose()
return _storage_service
requirements.txt
-3
View File
@@ -6,9 +6,6 @@ pydantic-settings==2.6.0
email-validator==2.2.0
python-multipart==0.0.32
# 邮件
aiosmtplib==3.0.2
# 测试
pytest==8.3.3
pytest-asyncio==0.24.0
scripts/init_production_env.sh
-1
View File
@@ -73,7 +73,6 @@ OSS_ACCESS_KEY_ID=$OSS_ACCESS_KEY_ID
OSS_ACCESS_KEY_SECRET=$OSS_ACCESS_KEY_SECRET
OSS_BUCKET_NAME=$OSS_BUCKET_NAME
LOG_LEVEL=INFO
CORS_ORIGINS_RAW=https://xiaoxiajianji.com,https://api.xiaoxiajianji.com
EOF
tests/integration/test_error_scenarios.py
+4 -6
View File
@@ -239,7 +239,7 @@ class TestForbidden403:
"""测试 403 禁止访问场景。"""
def test_access_other_user_project(self, auth_headers, other_auth_headers):
"""访问他人项目应返回 403/404（权限检查未实现时返回 200 为已知问题）。"""
"""访问他人项目应返回 403。"""
# 用户 A 创建项目
created = client.post(
"/api/v1/projects",
@@ -254,11 +254,10 @@ class TestForbidden403:
f"/api/v1/projects/{project_id}",
headers=other_auth_headers,
)
# TODO: 项目权限检查未实现，当前返回 200；实现后应改为 [403, 404]
assert response.status_code in [200, 403, 404], f"访问他人项目状态码异常: {response.status_code}"
assert response.status_code == 403, f"访问他人项目应返回 403，实际: {response.status_code}"
def test_delete_other_user_project(self, auth_headers, other_auth_headers):
"""删除他人项目应返回 403/404（权限检查未实现时返回 200/204 为已知问题）。"""
"""删除他人项目应返回 403。"""
created = client.post(
"/api/v1/projects",
json={"name": "Do Not Delete"},
@@ -271,8 +270,7 @@ class TestForbidden403:
f"/api/v1/projects/{project_id}",
headers=other_auth_headers,
)
# TODO: 项目权限检查未实现，当前可能返回 200/204；DELETE 端点未实现时返回 405；实现后应改为 [403, 404]
assert response.status_code in [200, 204, 403, 404, 405], f"删除他人项目状态码异常: {response.status_code}"
assert response.status_code == 403, f"删除他人项目应返回 403，实际: {response.status_code}"
# ---------------------------------------------------------------------------
tests/unit/test_audio_url_presign.py
+2 -2
View File
@@ -77,8 +77,8 @@ class TestAudioUrlSigner:
mock_svc = MagicMock(spec=OSSStorageService)
mock_svc.get_download_url.return_value = "https://signed/a.mp3?sig=123"
# 替换全局单例
with patch("app.core.storage._storage_service", mock_svc):
# 替换全局单例（现在位于 packages.shared.storage）
with patch("packages.shared.storage._storage_service", mock_svc):
from app.dependencies import get_audio_url_signer
signer = get_audio_url_signer()
tests/unit/test_p02_oss_credential_validation.py
+35 -35
View File
@@ -75,43 +75,43 @@ class TestOSSCredentialValidation:
class TestOSSDiagnose:
"""测试 OSSStorageService.diagnose() 方法。"""
"""测试 SharedStorageService.diagnose() 方法。"""
@patch("apps.api.app.core.storage.oss2", None)
@patch("apps.api.app.core.storage.get_settings")
@patch("packages.shared.storage.oss2", None)
@patch("packages.shared.storage.get_shared_settings")
def test_diagnose_logs_error_when_bucket_none(self, mock_settings, caplog):
"""bucket=None 时 diagnose 应输出 ERROR 日志。"""
from apps.api.app.core.storage import OSSStorageService
from packages.shared.storage import SharedStorageService
mock_settings.return_value.OSS_BUCKET_NAME = "test-bucket"
mock_settings.return_value.OSS_ENDPOINT = "oss-cn-test.com"
mock_settings.return_value.OSS_ACCESS_KEY_ID = ""
mock_settings.return_value.OSS_ACCESS_KEY_SECRET = ""
mock_settings.return_value.oss_bucket_name = "test-bucket"
mock_settings.return_value.oss_endpoint = "oss-cn-test.com"
mock_settings.return_value.oss_access_key_id = ""
mock_settings.return_value.oss_access_key_secret = ""
service = OSSStorageService()
service = SharedStorageService()
assert service.bucket is None
with caplog.at_level(logging.ERROR, logger="apps.api.app.core.storage"):
with caplog.at_level(logging.ERROR, logger="packages.shared.storage"):
service.diagnose()
assert any("❌" in record.message for record in caplog.records)
@patch("apps.api.app.core.storage.oss2")
@patch("apps.api.app.core.storage.get_settings")
@patch("packages.shared.storage.oss2")
@patch("packages.shared.storage.get_shared_settings")
def test_diagnose_logs_success_when_bucket_configured(self, mock_settings, mock_oss2, caplog):
"""bucket 已配置时 diagnose 应输出成功日志。"""
from apps.api.app.core.storage import OSSStorageService
from packages.shared.storage import SharedStorageService
mock_settings.return_value.OSS_BUCKET_NAME = "test-bucket"
mock_settings.return_value.OSS_ENDPOINT = "oss-cn-test.com"
mock_settings.return_value.OSS_ACCESS_KEY_ID = "test-key-id"
mock_settings.return_value.OSS_ACCESS_KEY_SECRET = "test-key-secret"
mock_settings.return_value.oss_bucket_name = "test-bucket"
mock_settings.return_value.oss_endpoint = "oss-cn-test.com"
mock_settings.return_value.oss_access_key_id = "test-key-id"
mock_settings.return_value.oss_access_key_secret = "test-key-secret"
mock_oss2.Bucket.return_value = MagicMock()
service = OSSStorageService()
service = SharedStorageService()
assert service.bucket is not None
with caplog.at_level(logging.INFO, logger="apps.api.app.core.storage"):
with caplog.at_level(logging.INFO, logger="packages.shared.storage"):
service.diagnose()
assert any("OSS诊断" in record.message for record in caplog.records)
@@ -120,38 +120,38 @@ class TestOSSDiagnose:
class TestOSSHTTPSEndpoint:
"""测试 P0-2 真正根因：sign_url 必须返回 HTTPS URL。"""
@patch("apps.api.app.core.storage.oss2")
@patch("apps.api.app.core.storage.get_settings")
@patch("packages.shared.storage.oss2")
@patch("packages.shared.storage.get_shared_settings")
def test_endpoint_without_scheme_gets_https_prefix(self, mock_settings, mock_oss2):
"""endpoint 无 scheme 时应自动加 https://，确保 sign_url 生成 HTTPS URL。"""
from apps.api.app.core.storage import OSSStorageService
from packages.shared.storage import SharedStorageService
mock_settings.return_value.OSS_BUCKET_NAME = "test-bucket"
mock_settings.return_value.OSS_ENDPOINT = "oss-cn-hangzhou.aliyuncs.com"
mock_settings.return_value.OSS_ACCESS_KEY_ID = "test-key-id"
mock_settings.return_value.OSS_ACCESS_KEY_SECRET = "test-key-secret"
mock_settings.return_value.oss_bucket_name = "test-bucket"
mock_settings.return_value.oss_endpoint = "oss-cn-hangzhou.aliyuncs.com"
mock_settings.return_value.oss_access_key_id = "test-key-id"
mock_settings.return_value.oss_access_key_secret = "test-key-secret"
mock_oss2.Bucket.return_value = MagicMock()
OSSStorageService()
SharedStorageService()
# 验证传给 oss2.Bucket 的 endpoint 带了 https://
call_args = mock_oss2.Bucket.call_args
endpoint_passed = call_args[0][1] # 第二个位置参数
assert endpoint_passed == "https://oss-cn-hangzhou.aliyuncs.com"
@patch("apps.api.app.core.storage.oss2")
@patch("apps.api.app.core.storage.get_settings")
@patch("packages.shared.storage.oss2")
@patch("packages.shared.storage.get_shared_settings")
def test_endpoint_with_existing_https_not_doubled(self, mock_settings, mock_oss2):
"""endpoint 已有 https:// 时不应重复添加。"""
from apps.api.app.core.storage import OSSStorageService
from packages.shared.storage import SharedStorageService
mock_settings.return_value.OSS_BUCKET_NAME = "test-bucket"
mock_settings.return_value.OSS_ENDPOINT = "https://oss-cn-hangzhou.aliyuncs.com"
mock_settings.return_value.OSS_ACCESS_KEY_ID = "test-key-id"
mock_settings.return_value.OSS_ACCESS_KEY_SECRET = "test-key-secret"
mock_settings.return_value.oss_bucket_name = "test-bucket"
mock_settings.return_value.oss_endpoint = "https://oss-cn-hangzhou.aliyuncs.com"
mock_settings.return_value.oss_access_key_id = "test-key-id"
mock_settings.return_value.oss_access_key_secret = "test-key-secret"
mock_oss2.Bucket.return_value = MagicMock()
OSSStorageService()
SharedStorageService()
call_args = mock_oss2.Bucket.call_args
endpoint_passed = call_args[0][1]