xiaoxia/xiaoxia-saas
1
0
Fork 0
Code Issues 15 Pull Requests 12 Actions 34.2k Packages Projects Releases 7 Wiki Activity

style: format tests/unit/test_p02_worker_oss_fix.py with black 26.5.1
CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Failing after 8s
Details
CI/CD Pipeline / Production Browser E2E (pull_request) Failing after 1609h45m0s
Details
CI/CD Pipeline / Staging API Integration Tests (pull_request) Failing after 1609h45m2s
Details
CI/CD Pipeline / Staging E2E Tests (pull_request) Failing after 1609h45m2s
Details
CI/CD Pipeline / Build Production Runtime Images (pull_request) Failing after 1609h45m4s
Details
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (pull_request) Failing after 1609h45m4s
Details
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
Details
CI/CD Pipeline / Deploy Production (pull_request) Failing after 1610h16m35s
Details

Browse Source
This commit is contained in:
xiaoxia
2026-07-11 13:41:12 +08:00
parent 8d90e8ea32
commit e01bfae30f
1 changed files with 104 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit/test_p02_worker_oss_fix.py
+104 -91
View File
@@ -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")