xiaoxia
841a168d30
CI/CD Pipeline / Validate Code Quality And Tests (push) Failing after 8s
CI/CD Pipeline / Frontend Lint (push) Successful in 4m7s
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
合并PR #216：给所有API返回的音频URL加OSS预签名 修复私有bucket下TTS合成音频、音色克隆参考音频、配音库音频等返回403的问题。
94 lines
3.3 KiB
Python
Executable File
94 lines
3.3 KiB
Python
Executable File
"""测试音频URL预签名逻辑。
|
|
|
|
验证所有 API 返回的音频 URL 都会经过 OSS 预签名（24小时有效期），
|
|
确保私有 bucket 下的音频文件前端可正常访问。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
|
|
class TestAudioUrlSigner:
|
|
"""测试音频URL签名函数的行为。"""
|
|
|
|
def _make_signer(self, mock_storage):
|
|
"""构造一个签名函数（模拟 get_audio_url_signer 的逻辑）。"""
|
|
|
|
def sign_audio_url(url: str) -> str:
|
|
if not url:
|
|
return url
|
|
return mock_storage.get_download_url(url, expires_seconds=86400)
|
|
|
|
return sign_audio_url
|
|
|
|
def test_empty_url_returns_empty(self):
|
|
"""空URL直接返回，不调用签名。"""
|
|
mock_storage = MagicMock()
|
|
signer = self._make_signer(mock_storage)
|
|
|
|
result = signer("")
|
|
assert result == ""
|
|
mock_storage.get_download_url.assert_not_called()
|
|
|
|
def test_none_url_returns_none(self):
|
|
"""None URL直接返回（有些字段可能为None）。"""
|
|
mock_storage = MagicMock()
|
|
signer = self._make_signer(mock_storage)
|
|
|
|
result = signer(None) # type: ignore
|
|
assert result is None
|
|
mock_storage.get_download_url.assert_not_called()
|
|
|
|
def test_valid_url_gets_signed_24h(self):
|
|
"""有效URL会调用 storage.get_download_url，有效期24小时（86400秒）。"""
|
|
mock_storage = MagicMock()
|
|
mock_storage.get_download_url.return_value = (
|
|
"https://bucket.oss-cn-hangzhou.aliyuncs.com/audio/test.mp3?signature=xxx"
|
|
)
|
|
signer = self._make_signer(mock_storage)
|
|
|
|
result = signer("https://bucket.oss-cn-hangzhou.aliyuncs.com/audio/test.mp3")
|
|
|
|
assert "signature=xxx" in result
|
|
mock_storage.get_download_url.assert_called_once_with(
|
|
"https://bucket.oss-cn-hangzhou.aliyuncs.com/audio/test.mp3",
|
|
expires_seconds=86400,
|
|
)
|
|
|
|
def test_storage_key_format_also_works(self):
|
|
"""纯 storage key 格式也能正常签名（storage内部会处理）。"""
|
|
mock_storage = MagicMock()
|
|
mock_storage.get_download_url.return_value = "https://signed-url/audio.mp3?sig=xxx"
|
|
signer = self._make_signer(mock_storage)
|
|
|
|
result = signer("audio/test.mp3")
|
|
|
|
assert result == "https://signed-url/audio.mp3?sig=xxx"
|
|
mock_storage.get_download_url.assert_called_once_with(
|
|
"audio/test.mp3",
|
|
expires_seconds=86400,
|
|
)
|
|
|
|
def test_signer_via_dependencies_module(self):
|
|
"""通过 dependencies 模块获取 signer，验证集成正确。"""
|
|
from app.core.storage import OSSStorageService
|
|
|
|
mock_svc = MagicMock(spec=OSSStorageService)
|
|
mock_svc.get_download_url.return_value = "https://signed/a.mp3?sig=123"
|
|
|
|
# 替换全局单例
|
|
with patch("app.core.storage._storage_service", mock_svc):
|
|
from app.dependencies import get_audio_url_signer
|
|
|
|
signer = get_audio_url_signer()
|
|
result = signer("test/audio.mp3")
|
|
|
|
assert result == "https://signed/a.mp3?sig=123"
|
|
mock_svc.get_download_url.assert_called_once_with(
|
|
"test/audio.mp3",
|
|
expires_seconds=86400,
|
|
)
|