From cefd97966aedd3cb6ecb96e33d526001f9ae54da Mon Sep 17 00:00:00 2001 From: CI Bot Date: Fri, 17 Jul 2026 15:33:18 +0800 Subject: [PATCH] =?UTF-8?q?fix(tests):=20=E4=BF=AE=E5=A4=8Doss=5Fhelpers?= =?UTF-8?q?=E6=B5=8B=E8=AF=95=20=E2=80=94=20=E6=94=B9=E7=94=A8patch=20oss?= =?UTF-8?q?=5Fsettings=E6=9B=BF=E4=BB=A3patch=20os.environ=20+=20black?= =?UTF-8?q?=E6=A0=BC=E5=BC=8F=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../worker/video_processing/render_adapter.py | 4 +- tests/unit/test_oss_upload_crash_fix.py | 60 +++------ tests/unit/test_p02_worker_oss_fix.py | 117 +++++------------- 3 files changed, 51 insertions(+), 130 deletions(-) diff --git a/apps/worker/video_processing/render_adapter.py b/apps/worker/video_processing/render_adapter.py index f74a01c73..46ca718c1 100755 --- a/apps/worker/video_processing/render_adapter.py +++ b/apps/worker/video_processing/render_adapter.py @@ -295,7 +295,9 @@ class RenderAdapter: except Exception: logger.exception("进度回调失败") - def _download_assets(self, clips: list[EditPlanClip], work_dir: Path) -> tuple[dict[str, Path], list[str], list[str]]: + def _download_assets( + self, clips: list[EditPlanClip], work_dir: Path + ) -> tuple[dict[str, Path], list[str], list[str]]: """下载片段素材到本地。 先通过 asset_id 批量查询 assets 表获取 file_url（OSS存储路径）， diff --git a/tests/unit/test_oss_upload_crash_fix.py b/tests/unit/test_oss_upload_crash_fix.py index 806e9b4cf..36d868b7f 100755 --- a/tests/unit/test_oss_upload_crash_fix.py +++ b/tests/unit/test_oss_upload_crash_fix.py @@ -9,7 +9,6 @@ from __future__ import annotations -import os import tempfile import time from pathlib import Path @@ -27,14 +26,9 @@ class TestOSSBucketConnectTimeout: mock_bucket_instance = MagicMock() with ( - patch.dict( - os.environ, - { - "OSS_ACCESS_KEY_ID": "test-key", - "OSS_ACCESS_KEY_SECRET": "test-secret", - "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com", - "OSS_BUCKET_NAME": "test-bucket", - }, + patch( + "video_processing.oss_helpers.oss_settings", + return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"), ), patch("video_processing.oss_helpers.oss2.Auth"), patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance) as mock_bucket_cls, @@ -53,7 +47,7 @@ class TestOSSBucketConnectTimeout: """OSS 配置缺失时返回 None.""" from video_processing.oss_helpers import oss_bucket - with patch.dict(os.environ, {}, clear=True): + with patch("video_processing.oss_helpers.oss_settings", return_value=None): bucket = oss_bucket() assert bucket is None @@ -80,14 +74,9 @@ class TestUploadToOSSMultipart: mock_bucket = MagicMock() with ( - patch.dict( - os.environ, - { - "OSS_ACCESS_KEY_ID": "test-key", - "OSS_ACCESS_KEY_SECRET": "test-secret", - "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com", - "OSS_BUCKET_NAME": "test-bucket", - }, + patch( + "video_processing.oss_helpers.oss_settings", + return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"), ), patch("video_processing.oss_helpers.oss2.Auth"), patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket), @@ -113,14 +102,9 @@ class TestUploadToOSSMultipart: mock_bucket = MagicMock() with ( - patch.dict( - os.environ, - { - "OSS_ACCESS_KEY_ID": "test-key", - "OSS_ACCESS_KEY_SECRET": "test-secret", - "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com", - "OSS_BUCKET_NAME": "test-bucket", - }, + patch( + "video_processing.oss_helpers.oss_settings", + return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"), ), patch("video_processing.oss_helpers.oss2.Auth"), patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket), @@ -167,14 +151,9 @@ class TestUploadToOSSTimeout: try: with ( - patch.dict( - os.environ, - { - "OSS_ACCESS_KEY_ID": "test-key", - "OSS_ACCESS_KEY_SECRET": "test-secret", - "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com", - "OSS_BUCKET_NAME": "test-bucket", - }, + patch( + "video_processing.oss_helpers.oss_settings", + return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"), ), patch("video_processing.oss_helpers.oss2.Auth"), patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket), @@ -200,14 +179,9 @@ class TestUploadToOSSTimeout: try: with ( - patch.dict( - os.environ, - { - "OSS_ACCESS_KEY_ID": "test-key", - "OSS_ACCESS_KEY_SECRET": "test-secret", - "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com", - "OSS_BUCKET_NAME": "test-bucket", - }, + patch( + "video_processing.oss_helpers.oss_settings", + return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"), ), patch("video_processing.oss_helpers.oss2.Auth"), patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket), @@ -227,7 +201,7 @@ class TestUploadToOSSTimeout: file_path = Path(small_file.name) try: - with patch.dict(os.environ, {}, clear=True): + with patch("video_processing.oss_helpers.oss_settings", return_value=None): url = upload_to_oss(file_path, "test/noconfig.mp4") assert url is None finally: diff --git a/tests/unit/test_p02_worker_oss_fix.py b/tests/unit/test_p02_worker_oss_fix.py index 58f7d37fd..f9c12826b 100755 --- a/tests/unit/test_p02_worker_oss_fix.py +++ b/tests/unit/test_p02_worker_oss_fix.py @@ -8,7 +8,7 @@ from __future__ import annotations -import os +from pathlib import Path from unittest.mock import MagicMock, patch # ── oss_bucket endpoint scheme 修复 ────────────────────────────────────────── @@ -23,20 +23,13 @@ class TestOSSBucketEndpointScheme: mock_bucket_instance = MagicMock() with ( - patch.dict( - os.environ, - { - "OSS_ACCESS_KEY_ID": "test-key", - "OSS_ACCESS_KEY_SECRET": "test-secret", - "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com", - "OSS_BUCKET_NAME": "test-bucket", - }, + patch( + "video_processing.oss_helpers.oss_settings", + return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"), ), patch("video_processing.oss_helpers.oss2.Auth"), patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance) as mock_bucket_cls, ): - # 清除缓存，确保重新创建 - bucket = oss_bucket() assert bucket is mock_bucket_instance @@ -52,19 +45,13 @@ class TestOSSBucketEndpointScheme: mock_bucket_instance = MagicMock() with ( - patch.dict( - os.environ, - { - "OSS_ACCESS_KEY_ID": "test-key", - "OSS_ACCESS_KEY_SECRET": "test-secret", - "OSS_ENDPOINT": "https://oss-cn-hangzhou.aliyuncs.com", - "OSS_BUCKET_NAME": "test-bucket", - }, + patch( + "video_processing.oss_helpers.oss_settings", + return_value=("test-key", "test-secret", "https://oss-cn-hangzhou.aliyuncs.com", "test-bucket"), ), patch("video_processing.oss_helpers.oss2.Auth"), patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance) as mock_bucket_cls, ): - oss_bucket() call_args = mock_bucket_cls.call_args @@ -79,19 +66,13 @@ class TestOSSBucketEndpointScheme: mock_bucket_instance = MagicMock() with ( - patch.dict( - os.environ, - { - "OSS_ACCESS_KEY_ID": "test-key", - "OSS_ACCESS_KEY_SECRET": "test-secret", - "OSS_ENDPOINT": "http://oss-cn-hangzhou.aliyuncs.com", - "OSS_BUCKET_NAME": "test-bucket", - }, + patch( + "video_processing.oss_helpers.oss_settings", + return_value=("test-key", "test-secret", "http://oss-cn-hangzhou.aliyuncs.com", "test-bucket"), ), patch("video_processing.oss_helpers.oss2.Auth"), patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance) as mock_bucket_cls, ): - oss_bucket() call_args = mock_bucket_cls.call_args @@ -102,17 +83,7 @@ class TestOSSBucketEndpointScheme: """凭证缺失时返回 None.""" from video_processing.oss_helpers import oss_bucket - with patch.dict( - os.environ, - { - "OSS_ACCESS_KEY_ID": "", - "OSS_ACCESS_KEY_SECRET": "", - "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com", - "OSS_BUCKET_NAME": "test-bucket", - }, - clear=True, - ): - + with patch("video_processing.oss_helpers.oss_settings", return_value=None): bucket = oss_bucket() assert bucket is None @@ -128,17 +99,15 @@ class TestGetSignedDownloadUrl: from video_processing.oss_helpers import get_signed_download_url mock_bucket = MagicMock() - mock_bucket.sign_url.return_value = "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/test.mp4?OSSAccessKeyId=xxx&Expires=xxx&Signature=xxx" + mock_bucket.sign_url.return_value = ( + "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/test.mp4" + "?OSSAccessKeyId=xxx&Expires=xxx&Signature=xxx" + ) with ( - patch.dict( - os.environ, - { - "OSS_ACCESS_KEY_ID": "test-key", - "OSS_ACCESS_KEY_SECRET": "test-secret", - "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com", - "OSS_BUCKET_NAME": "test-bucket", - }, + patch( + "video_processing.oss_helpers.oss_settings", + return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"), ), patch("video_processing.oss_helpers.oss2.Auth"), patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket), @@ -159,14 +128,9 @@ class TestGetSignedDownloadUrl: ) with ( - patch.dict( - os.environ, - { - "OSS_ACCESS_KEY_ID": "test-key", - "OSS_ACCESS_KEY_SECRET": "test-secret", - "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com", - "OSS_BUCKET_NAME": "test-bucket", - }, + patch( + "video_processing.oss_helpers.oss_settings", + return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"), ), patch("video_processing.oss_helpers.oss2.Auth"), patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket), @@ -183,7 +147,7 @@ class TestGetSignedDownloadUrl: """bucket 为 None 时返回 None（不抛异常）.""" from video_processing.oss_helpers import get_signed_download_url - with patch.dict(os.environ, {}, clear=True): + with patch("video_processing.oss_helpers.oss_settings", return_value=None): result = get_signed_download_url("generated/test.mp4") assert result is None @@ -195,14 +159,9 @@ class TestGetSignedDownloadUrl: mock_bucket.sign_url.side_effect = Exception("sign failed") with ( - patch.dict( - os.environ, - { - "OSS_ACCESS_KEY_ID": "test-key", - "OSS_ACCESS_KEY_SECRET": "test-secret", - "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com", - "OSS_BUCKET_NAME": "test-bucket", - }, + patch( + "video_processing.oss_helpers.oss_settings", + return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"), ), patch("video_processing.oss_helpers.oss2.Auth"), patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket), @@ -224,17 +183,10 @@ class TestUploadToOSSReturnsHTTPS: mock_bucket = MagicMock() mock_bucket.put_object_from_file = MagicMock() - from pathlib import Path - with ( - patch.dict( - os.environ, - { - "OSS_ACCESS_KEY_ID": "test-key", - "OSS_ACCESS_KEY_SECRET": "test-secret", - "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com", - "OSS_BUCKET_NAME": "test-bucket", - }, + patch( + "video_processing.oss_helpers.oss_settings", + return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"), ), patch("video_processing.oss_helpers.oss2.Auth"), patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket), @@ -252,17 +204,10 @@ class TestUploadToOSSReturnsHTTPS: mock_bucket = MagicMock() mock_bucket.put_object_from_file = MagicMock() - from pathlib import Path - with ( - patch.dict( - os.environ, - { - "OSS_ACCESS_KEY_ID": "test-key", - "OSS_ACCESS_KEY_SECRET": "test-secret", - "OSS_ENDPOINT": "https://oss-cn-hangzhou.aliyuncs.com", - "OSS_BUCKET_NAME": "test-bucket", - }, + patch( + "video_processing.oss_helpers.oss_settings", + return_value=("test-key", "test-secret", "https://oss-cn-hangzhou.aliyuncs.com", "test-bucket"), ), patch("video_processing.oss_helpers.oss2.Auth"), patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),