diff --git a/.env.production.example b/.env.production.example index 43cb2312d..5e2c6898b 100755 --- a/.env.production.example +++ b/.env.production.example @@ -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 diff --git a/apps/api/app/config.py b/apps/api/app/config.py index 505281bd2..68ec61b5d 100755 --- a/apps/api/app/config.py +++ b/apps/api/app/config.py @@ -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 diff --git a/apps/api/app/core/storage.py b/apps/api/app/core/storage.py index df064c86d..8c39c13f8 100644 --- a/apps/api/app/core/storage.py +++ b/apps/api/app/core/storage.py @@ -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"] diff --git a/packages/shared/storage.py b/packages/shared/storage.py index 28205d5f4..0c7aa8ce3 100644 --- a/packages/shared/storage.py +++ b/packages/shared/storage.py @@ -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 diff --git a/requirements.txt b/requirements.txt index cda4fe188..7b9fbf702 100755 --- a/requirements.txt +++ b/requirements.txt @@ -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 diff --git a/scripts/init_production_env.sh b/scripts/init_production_env.sh index 32d376ee2..2da9bc192 100755 --- a/scripts/init_production_env.sh +++ b/scripts/init_production_env.sh @@ -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 diff --git a/tests/integration/test_error_scenarios.py b/tests/integration/test_error_scenarios.py index f11e36504..34601720e 100755 --- a/tests/integration/test_error_scenarios.py +++ b/tests/integration/test_error_scenarios.py @@ -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}" # --------------------------------------------------------------------------- diff --git a/tests/unit/test_audio_url_presign.py b/tests/unit/test_audio_url_presign.py index 75f8f08f2..b26f964be 100755 --- a/tests/unit/test_audio_url_presign.py +++ b/tests/unit/test_audio_url_presign.py @@ -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() diff --git a/tests/unit/test_p02_oss_credential_validation.py b/tests/unit/test_p02_oss_credential_validation.py index deaca7dcc..abf67590d 100644 --- a/tests/unit/test_p02_oss_credential_validation.py +++ b/tests/unit/test_p02_oss_credential_validation.py @@ -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]