From e01bfae30f3a0bbe72e785faf5aaf6f82687a990 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Sat, 11 Jul 2026 13:41:12 +0800 Subject: [PATCH] style: format tests/unit/test_p02_worker_oss_fix.py with black 26.5.1 --- tests/unit/test_p02_worker_oss_fix.py | 195 ++++++++++++++------------ 1 file changed, 104 insertions(+), 91 deletions(-) diff --git a/tests/unit/test_p02_worker_oss_fix.py b/tests/unit/test_p02_worker_oss_fix.py index 3be633bfc..f76d3679c 100755 --- a/tests/unit/test_p02_worker_oss_fix.py +++ b/tests/unit/test_p02_worker_oss_fix.py @@ -13,7 +13,6 @@ from unittest.mock import MagicMock, patch import pytest - # ── oss_bucket endpoint scheme 修复 ────────────────────────────────────────── @@ -25,17 +24,19 @@ class TestOSSBucketEndpointScheme: from video_processing.oss_helpers import oss_bucket 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.oss2.Auth") as mock_auth, patch( - "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance - ) as mock_bucket_cls: + 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.oss2.Auth") as mock_auth, + patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance) as mock_bucket_cls, + ): # 清除缓存，确保重新创建 import video_processing.oss_helpers as oss_mod @@ -45,9 +46,7 @@ class TestOSSBucketEndpointScheme: # 验证 endpoint 传的是带 https:// 的 call_args = mock_bucket_cls.call_args endpoint_arg = call_args[0][1] # 第 2 个位置参数是 endpoint - assert endpoint_arg.startswith("https://"), ( - f"endpoint 应该带 https:// 前缀，实际为: {endpoint_arg}" - ) + assert endpoint_arg.startswith("https://"), f"endpoint 应该带 https:// 前缀，实际为: {endpoint_arg}" assert "oss-cn-hangzhou.aliyuncs.com" in endpoint_arg def test_endpoint_with_https_keeps_as_is(self): @@ -55,17 +54,19 @@ class TestOSSBucketEndpointScheme: from video_processing.oss_helpers import oss_bucket 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.oss2.Auth"), patch( - "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance - ) as mock_bucket_cls: + 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.oss2.Auth"), + patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance) as mock_bucket_cls, + ): import video_processing.oss_helpers as oss_mod bucket = oss_bucket() @@ -81,17 +82,19 @@ class TestOSSBucketEndpointScheme: from video_processing.oss_helpers import oss_bucket 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.oss2.Auth"), patch( - "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance - ) as mock_bucket_cls: + 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.oss2.Auth"), + patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance) as mock_bucket_cls, + ): import video_processing.oss_helpers as oss_mod bucket = oss_bucket() @@ -133,16 +136,18 @@ class TestGetSignedDownloadUrl: 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" - 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.oss2.Auth"), patch( - "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket + 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.oss2.Auth"), + patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket), ): result = get_signed_download_url("generated/test.mp4", expires_seconds=3600) @@ -155,22 +160,24 @@ 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?sign=xxx" + mock_bucket.sign_url.return_value = ( + "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/test.mp4?sign=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.oss2.Auth"), patch( - "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket + 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.oss2.Auth"), + patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket), ): - result = get_signed_download_url( - "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/test.mp4" - ) + result = get_signed_download_url("https://test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/test.mp4") mock_bucket.sign_url.assert_called_once() # 验证传给 sign_url 的是纯 storage key，不是完整 URL @@ -193,16 +200,18 @@ class TestGetSignedDownloadUrl: mock_bucket = MagicMock() 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.oss2.Auth"), patch( - "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket + 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.oss2.Auth"), + patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket), ): result = get_signed_download_url("generated/test.mp4") assert result is None @@ -223,16 +232,18 @@ class TestUploadToOSSReturnsHTTPS: 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.oss2.Auth"), patch( - "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket + 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.oss2.Auth"), + patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket), ): result = upload_to_oss(Path("/tmp/test.mp4"), "generated/test.mp4") @@ -249,16 +260,18 @@ class TestUploadToOSSReturnsHTTPS: 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.oss2.Auth"), patch( - "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket + 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.oss2.Auth"), + patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket), ): result = upload_to_oss(Path("/tmp/test.mp4"), "generated/test.mp4")