xiaoxia
fe2ab121e7
CI/CD Pipeline / Validate Code Quality And Tests (push) Failing after 40s
CI/CD Pipeline / Unit Tests (push) Failing after 1m42s
CI/CD Pipeline / Integration Tests (push) Successful in 1m20s
CI/CD Pipeline / Frontend Lint (push) Has been cancelled
CI/CD Pipeline / Build Staging API Image (push) Has been cancelled
CI/CD Pipeline / Build Staging Worker Image (push) Has been cancelled
CI/CD Pipeline / Build Staging Web Image (push) Has been cancelled
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Has been cancelled
CI/CD Pipeline / Staging E2E Tests (push) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (push) Has been cancelled
CI/CD Pipeline / Build Production API Image (push) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (push) Has been cancelled
CI/CD Pipeline / Build Production Web Image (push) Has been cancelled
CI/CD Pipeline / Deploy Production (push) Has been cancelled
CI/CD Pipeline / Production Browser E2E (push) Has been cancelled
fix(security): 后端安全技术债务第二轮（P1+P2+P2） - audio_merger裸subprocess下沉 + ffmpeg_utils架构下沉到packages/shared - _verify_url_accessible重定向每跳SSRF校验 - url_security下载文件魔数校验 - 新增33个单测
267 lines
10 KiB
Python
267 lines
10 KiB
Python
"""P0/P1 修复单元测试 — 一键生成 P0 问题 + P1 校验.
|
||
|
||
覆盖：
|
||
P0-1: _download_library_assets 双模式查询（asset_library_id / project_id）
|
||
P0-2: OSS 上传失败抛异常 + URL 可访问性校验
|
||
P0-3: FFmpeg 失败时完整 stderr 日志
|
||
P1: template_id 存在性校验 + asset_ids 归属校验
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import subprocess
|
||
import sys
|
||
from pathlib import Path
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
import pytest
|
||
|
||
# 添加 worker app 到 sys.path
|
||
_WORKER_ROOT = Path(__file__).resolve().parents[2] / "apps" / "worker"
|
||
if str(_WORKER_ROOT) not in sys.path:
|
||
sys.path.insert(0, str(_WORKER_ROOT))
|
||
|
||
|
||
# ── P0-1: _download_library_assets ────────────────────────────────────────────
|
||
|
||
|
||
class TestDownloadLibraryAssets:
|
||
"""P0-1: 素材下载双模式 + 错误处理."""
|
||
|
||
def _make_asset(self, id_: str, file_url: str, project_id: str = "p1", library_id: str = "lib1"):
|
||
mock = MagicMock()
|
||
mock.id = id_
|
||
mock.file_url = file_url
|
||
mock.name = f"asset_{id_}"
|
||
mock.project_id = project_id
|
||
mock.asset_library_id = library_id
|
||
return mock
|
||
|
||
@patch("worker_app.tasks.generation.SessionLocal")
|
||
@patch("worker_app.tasks.generation.download_asset")
|
||
def test_asset_library_mode(self, mock_download, mock_session_factory):
|
||
"""素材库模式：按 asset_library_id 查询."""
|
||
from worker_app.tasks.generation import _download_library_assets
|
||
|
||
session = MagicMock()
|
||
mock_session_factory.return_value = session
|
||
query = MagicMock()
|
||
session.query.return_value = query
|
||
filter_result = MagicMock()
|
||
query.filter.return_value = filter_result
|
||
in_filter = MagicMock()
|
||
filter_result.filter.return_value = in_filter
|
||
assets = [self._make_asset("a1", "video/a1.mp4")]
|
||
in_filter.order_by.return_value.all.return_value = assets
|
||
|
||
mock_download.return_value = True
|
||
|
||
with patch("worker_app.tasks.generation.AssetModel", create=True):
|
||
result = _download_library_assets(
|
||
Path("/tmp/test"),
|
||
asset_library_id="lib1",
|
||
)
|
||
|
||
assert len(result) == 1
|
||
mock_download.assert_called_once()
|
||
|
||
@patch("worker_app.tasks.generation.SessionLocal")
|
||
@patch("worker_app.tasks.generation.download_asset")
|
||
def test_project_mode(self, mock_download, mock_session_factory):
|
||
"""项目级模式：asset_library_id 为空时按 project_id 查询."""
|
||
from worker_app.tasks.generation import _download_library_assets
|
||
|
||
session = MagicMock()
|
||
mock_session_factory.return_value = session
|
||
query = MagicMock()
|
||
session.query.return_value = query
|
||
filter_result = MagicMock()
|
||
query.filter.return_value = filter_result
|
||
proj_filter = MagicMock()
|
||
filter_result.filter.return_value = proj_filter
|
||
assets = [self._make_asset("a1", "video/a1.mp4", project_id="proj1")]
|
||
proj_filter.order_by.return_value.all.return_value = assets
|
||
|
||
mock_download.return_value = True
|
||
|
||
result = _download_library_assets(
|
||
Path("/tmp/test"),
|
||
project_id="proj1",
|
||
)
|
||
|
||
assert len(result) == 1
|
||
|
||
def test_both_empty_raises(self):
|
||
"""asset_library_id 和 project_id 都为空时抛 ValueError."""
|
||
from worker_app.tasks.generation import _download_library_assets
|
||
|
||
with pytest.raises(ValueError, match="至少需要提供一个"):
|
||
_download_library_assets(Path("/tmp/test"))
|
||
|
||
@patch("worker_app.tasks.generation.SessionLocal")
|
||
def test_no_assets_found_raises(self, mock_session_factory):
|
||
"""查不到素材时抛 RuntimeError."""
|
||
from worker_app.tasks.generation import _download_library_assets
|
||
|
||
session = MagicMock()
|
||
mock_session_factory.return_value = session
|
||
query = MagicMock()
|
||
session.query.return_value = query
|
||
filter_result = MagicMock()
|
||
query.filter.return_value = filter_result
|
||
in_filter = MagicMock()
|
||
filter_result.filter.return_value = in_filter
|
||
in_filter.order_by.return_value.all.return_value = []
|
||
|
||
with pytest.raises(RuntimeError, match="未找到视频素材"):
|
||
_download_library_assets(
|
||
Path("/tmp/test"),
|
||
asset_library_id="lib1",
|
||
)
|
||
|
||
@patch("worker_app.tasks.generation.SessionLocal")
|
||
@patch("worker_app.tasks.generation.download_asset")
|
||
def test_all_asset_ids_fail_raises(self, mock_download, mock_session_factory):
|
||
"""指定 asset_ids 但全部下载失败时抛 RuntimeError."""
|
||
from worker_app.tasks.generation import _download_library_assets
|
||
|
||
session = MagicMock()
|
||
mock_session_factory.return_value = session
|
||
query = MagicMock()
|
||
session.query.return_value = query
|
||
filter_result = MagicMock()
|
||
query.filter.return_value = filter_result
|
||
id_filter = MagicMock()
|
||
filter_result.filter.return_value = id_filter
|
||
assets = [self._make_asset("a1", "video/a1.mp4")]
|
||
id_filter.order_by.return_value.all.return_value = assets
|
||
|
||
mock_download.return_value = False # 全部下载失败
|
||
|
||
with pytest.raises(RuntimeError, match="素材下载失败"):
|
||
_download_library_assets(
|
||
Path("/tmp/test"),
|
||
asset_library_id="lib1",
|
||
asset_ids=["a1"],
|
||
)
|
||
|
||
|
||
# ── P0-2: OSS 上传 + URL 校验 ────────────────────────────────────────────────
|
||
|
||
|
||
class TestOSSUploadAndVerify:
|
||
"""P0-2: OSS 上传失败抛异常 + URL 可访问性校验."""
|
||
|
||
def test_verify_url_accessible_success(self):
|
||
"""URL 可访问时返回 True."""
|
||
from worker_app.tasks.generation import _verify_url_accessible
|
||
|
||
mock_response = MagicMock()
|
||
mock_response.status = 200
|
||
mock_response.__enter__ = MagicMock(return_value=mock_response)
|
||
mock_response.__exit__ = MagicMock(return_value=False)
|
||
|
||
with patch("urllib.request.OpenerDirector.open", return_value=mock_response):
|
||
assert _verify_url_accessible("https://example.com/test.mp4") is True
|
||
|
||
def test_verify_url_accessible_failure(self):
|
||
"""URL 不可访问时返回 False."""
|
||
from worker_app.tasks.generation import _verify_url_accessible
|
||
|
||
with patch("urllib.request.OpenerDirector.open", side_effect=Exception("connection refused")):
|
||
assert _verify_url_accessible("https://example.com/test.mp4") is False
|
||
|
||
def test_verify_url_404(self):
|
||
"""URL 返回 404 时返回 False."""
|
||
from worker_app.tasks.generation import _verify_url_accessible
|
||
|
||
mock_response = MagicMock()
|
||
mock_response.status = 404
|
||
mock_response.__enter__ = MagicMock(return_value=mock_response)
|
||
mock_response.__exit__ = MagicMock(return_value=False)
|
||
|
||
with patch("urllib.request.OpenerDirector.open", return_value=mock_response):
|
||
assert _verify_url_accessible("https://example.com/test.mp4") is False
|
||
|
||
|
||
# ── P0-3: FFmpeg stderr 日志 ─────────────────────────────────────────────────
|
||
|
||
|
||
class TestFFmpegStderrLogging:
|
||
"""P0-3: FFmpeg 失败时完整 stderr 打到日志."""
|
||
|
||
def test_run_ffmpeg_logs_stderr_on_failure(self, caplog):
|
||
"""run_ffmpeg 失败时记录 stderr 到日志."""
|
||
from video_processing.ffmpeg_utils import run_ffmpeg
|
||
|
||
error = subprocess.CalledProcessError(
|
||
returncode=183,
|
||
cmd=["ffmpeg", "-y", "-i", "input.mp4", "output.mp4"],
|
||
output="",
|
||
stderr="Error message from ffmpeg: filter graph error details here",
|
||
)
|
||
|
||
with patch("subprocess.run", side_effect=error):
|
||
with caplog.at_level(logging.ERROR):
|
||
with pytest.raises(subprocess.CalledProcessError):
|
||
run_ffmpeg(["ffmpeg", "-y", "-i", "input.mp4", "output.mp4"])
|
||
|
||
assert "FFmpeg 命令失败" in caplog.text
|
||
assert "exit_code=183" in caplog.text
|
||
assert "filter graph error" in caplog.text
|
||
|
||
def test_run_ffmpeg_success(self):
|
||
"""run_ffmpeg 成功时正常返回."""
|
||
from video_processing.ffmpeg_utils import run_ffmpeg
|
||
|
||
mock_result = MagicMock()
|
||
mock_result.stdout = "output"
|
||
mock_result.stderr = ""
|
||
|
||
with patch("subprocess.run", return_value=mock_result):
|
||
stdout, stderr = run_ffmpeg(["ffmpeg", "-version"])
|
||
assert stdout == "output"
|
||
assert stderr == ""
|
||
|
||
|
||
# ── P1: template_id + asset_ids 校验 ─────────────────────────────────────────
|
||
|
||
|
||
class TestP1Validations:
|
||
"""P1: template_id 存在性校验 + asset_ids 归属校验."""
|
||
|
||
def test_validate_template_exists_success(self):
|
||
"""模板存在时不抛异常."""
|
||
from worker_app.tasks.generation import _validate_template_exists
|
||
|
||
mock_template = MagicMock()
|
||
mock_template.id = "tmpl_001"
|
||
mock_template.name = "Test Template"
|
||
mock_template.is_active = True
|
||
|
||
session = MagicMock()
|
||
mock_session = MagicMock()
|
||
session.query.return_value = mock_session
|
||
filter_result = MagicMock()
|
||
mock_session.filter.return_value = filter_result
|
||
filter_result.first.return_value = mock_template
|
||
|
||
with patch("worker_app.tasks.generation.SessionLocal", return_value=session):
|
||
_validate_template_exists("tmpl_001") # 不抛异常
|
||
|
||
def test_validate_template_exists_not_found(self):
|
||
"""模板不存在时抛 ValueError."""
|
||
from worker_app.tasks.generation import _validate_template_exists
|
||
|
||
session = MagicMock()
|
||
mock_session = MagicMock()
|
||
session.query.return_value = mock_session
|
||
filter_result = MagicMock()
|
||
mock_session.filter.return_value = filter_result
|
||
filter_result.first.return_value = None
|
||
|
||
with patch("worker_app.tasks.generation.SessionLocal", return_value=session):
|
||
with pytest.raises(ValueError, match="模板不存在"):
|
||
_validate_template_exists("tmpl_nonexistent")
|