xiaoxia
48e5077191
CI/CD Pipeline / Validate Code Quality And Tests (push) Failing after 9s
CI/CD Pipeline / Frontend Lint (push) Successful in 1m31s
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (push) Has been skipped
CI/CD Pipeline / Build Production Runtime Images (push) Has been skipped
CI/CD Pipeline / Staging E2E Tests (push) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
269 lines
11 KiB
Python
Executable File
269 lines
11 KiB
Python
Executable File
"""P0-2 深度修复：Worker 端 OSS 工具函数测试.
|
|
|
|
测试：
|
|
1. oss_bucket() endpoint 自动补 https:// 前缀
|
|
2. get_signed_download_url() 生成预签名 URL
|
|
3. upload_to_oss() 返回 HTTPS URL
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
|
|
# ── oss_bucket endpoint scheme 修复 ──────────────────────────────────────────
|
|
|
|
|
|
class TestOSSBucketEndpointScheme:
|
|
"""测试 oss_bucket() 自动为 endpoint 补 https:// 前缀."""
|
|
|
|
def test_endpoint_without_scheme_adds_https(self):
|
|
"""endpoint 不带 scheme 时，自动补 https://."""
|
|
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:
|
|
# 清除缓存，确保重新创建
|
|
import video_processing.oss_helpers as oss_mod
|
|
|
|
bucket = oss_bucket()
|
|
|
|
assert bucket is mock_bucket_instance
|
|
# 验证 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 "oss-cn-hangzhou.aliyuncs.com" in endpoint_arg
|
|
|
|
def test_endpoint_with_https_keeps_as_is(self):
|
|
"""endpoint 已有 https:// 时，不重复添加."""
|
|
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:
|
|
import video_processing.oss_helpers as oss_mod
|
|
|
|
bucket = oss_bucket()
|
|
|
|
call_args = mock_bucket_cls.call_args
|
|
endpoint_arg = call_args[0][1]
|
|
# 不应该出现 https://https:// 这种双重前缀
|
|
assert endpoint_arg.count("https://") == 1
|
|
assert endpoint_arg == "https://oss-cn-hangzhou.aliyuncs.com"
|
|
|
|
def test_endpoint_with_http_keeps_as_is(self):
|
|
"""endpoint 已有 http:// 时，不修改（保留用户选择）."""
|
|
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:
|
|
import video_processing.oss_helpers as oss_mod
|
|
|
|
bucket = oss_bucket()
|
|
|
|
call_args = mock_bucket_cls.call_args
|
|
endpoint_arg = call_args[0][1]
|
|
assert endpoint_arg == "http://oss-cn-hangzhou.aliyuncs.com"
|
|
|
|
def test_missing_credentials_returns_none(self):
|
|
"""凭证缺失时返回 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,
|
|
):
|
|
import video_processing.oss_helpers as oss_mod
|
|
|
|
bucket = oss_bucket()
|
|
assert bucket is None
|
|
|
|
|
|
# ── get_signed_download_url ──────────────────────────────────────────────────
|
|
|
|
|
|
class TestGetSignedDownloadUrl:
|
|
"""测试 get_signed_download_url() 预签名 URL 生成."""
|
|
|
|
def test_returns_signed_url_with_storage_key(self):
|
|
"""传入 storage key 时，调用 sign_url 并返回结果."""
|
|
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"
|
|
|
|
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)
|
|
|
|
assert result is not None
|
|
assert "Signature=" in result
|
|
mock_bucket.sign_url.assert_called_once_with("GET", "generated/test.mp4", 3600)
|
|
|
|
def test_normalizes_full_url_to_storage_key(self):
|
|
"""传入完整 URL 时，提取 storage key 再生成签名."""
|
|
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"
|
|
|
|
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"
|
|
)
|
|
|
|
mock_bucket.sign_url.assert_called_once()
|
|
# 验证传给 sign_url 的是纯 storage key，不是完整 URL
|
|
call_key = mock_bucket.sign_url.call_args[0][1]
|
|
assert not call_key.startswith("http")
|
|
assert call_key == "generated/test.mp4"
|
|
|
|
def test_returns_none_when_bucket_none(self):
|
|
"""bucket 为 None 时返回 None（不抛异常）."""
|
|
from video_processing.oss_helpers import get_signed_download_url
|
|
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
result = get_signed_download_url("generated/test.mp4")
|
|
assert result is None
|
|
|
|
def test_sign_url_exception_returns_none(self):
|
|
"""sign_url 抛异常时，返回 None（不向上抛出）."""
|
|
from video_processing.oss_helpers import get_signed_download_url
|
|
|
|
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
|
|
):
|
|
result = get_signed_download_url("generated/test.mp4")
|
|
assert result is None
|
|
|
|
|
|
# ── upload_to_oss 返回 HTTPS URL ─────────────────────────────────────────────
|
|
|
|
|
|
class TestUploadToOSSReturnsHTTPS:
|
|
"""测试 upload_to_oss() 返回的 URL 始终是 HTTPS."""
|
|
|
|
def test_endpoint_without_scheme_returns_https_url(self):
|
|
"""endpoint 不带 scheme 时，返回 HTTPS URL."""
|
|
from video_processing.oss_helpers import upload_to_oss
|
|
|
|
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.oss2.Auth"), patch(
|
|
"video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket
|
|
):
|
|
result = upload_to_oss(Path("/tmp/test.mp4"), "generated/test.mp4")
|
|
|
|
assert result is not None
|
|
assert result.startswith("https://")
|
|
assert "test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/test.mp4" in result
|
|
|
|
def test_endpoint_with_https_returns_clean_url(self):
|
|
"""endpoint 带 https:// 时，URL 里不会有双重 https."""
|
|
from video_processing.oss_helpers import upload_to_oss
|
|
|
|
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.oss2.Auth"), patch(
|
|
"video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket
|
|
):
|
|
result = upload_to_oss(Path("/tmp/test.mp4"), "generated/test.mp4")
|
|
|
|
assert result is not None
|
|
assert result.startswith("https://")
|
|
# 不应该出现 https://https://
|
|
assert result.count("https://") == 1
|