CI Bot
1a57878f76
CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Failing after 1m12s
CI/CD Pipeline / Unit Tests (pull_request) Failing after 2m17s
CI/CD Pipeline / Integration Tests (pull_request) Failing after 1m33s
CI/CD Pipeline / Frontend Lint (pull_request) Successful in 3m14s
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Build Production Runtime Images (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
1. 未使用依赖清理:
- 从 requirements-base.txt 移除 cryptography 和 pyOpenSSL
2. pyflakes 警告清零 (apps/ + packages/ + tests/):
- 移除 17 处未使用的 import (F401)
- 修复 26 处未使用的局部变量 (F841):
* 有副作用的赋值转为裸调用
* 无副作用的赋值直接删除
- 修复 1 处未使用的异常变量 (F841)
- 修复 1 处空 except 块
3. 测试文件冗余清理:
- 删除 tests/integration/test_project_management.py (模块级 skip，测试不存在的模块)
- 删除 tests/integration/fixtures/duplication_routes_fixed.py (未被引用)
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
276 lines
11 KiB
Python
Executable File
276 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
|
|
|
|
# ── 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"),
|
|
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
|
|
# 验证 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,
|
|
):
|
|
|
|
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,
|
|
):
|
|
|
|
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,
|
|
):
|
|
|
|
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),
|
|
):
|
|
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
|