CI Bot
aca20d9bf0
CI/CD Pipeline / Frontend Lint (pull_request) Successful in 3m28s
CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Successful in 6m18s
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 / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / Integration Tests (pull_request) Successful in 2m0s
根因：新引擎 UnifiedRenderService 所有 FFmpeg 调用通过 run_ffmpeg 执行， 但 subprocess.run 未设置 timeout，FFmpeg hang 住时 worker 线程永久阻塞。 旧引擎 compose_video 有单独的 timeout=3600，但新引擎路径没有。 修复： 1. run_ffmpeg 新增默认超时 1800s（30分钟），支持自定义传参 2. 捕获 TimeoutExpired 并打 error 日志后重新抛出 3. probe_video_info 新增 timeout=15s 超时保护 4. 7个单元测试覆盖超时逻辑 影响范围：所有通过 run_ffmpeg 调用的 FFmpeg 命令 （unified_render_service 所有渲染/混音/合并操作）
94 lines
4.0 KiB
Python
94 lines
4.0 KiB
Python
"""FFmpeg 超时保护测试。
|
|
|
|
验证 run_ffmpeg / probe_video_info 的超时保护机制，
|
|
防止 FFmpeg hang 住导致 worker 永久阻塞。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import subprocess
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from video_processing.ffmpeg_utils import (
|
|
DEFAULT_FFMPEG_TIMEOUT,
|
|
probe_video_info,
|
|
run_ffmpeg,
|
|
)
|
|
|
|
# ── run_ffmpeg 超时保护 ──────────────────────────────────────────────────────
|
|
|
|
|
|
class TestRunFFmpegTimeout:
|
|
"""run_ffmpeg 超时保护测试。"""
|
|
|
|
def test_default_timeout_is_set(self):
|
|
"""默认超时应为 1800 秒（30分钟）。"""
|
|
assert DEFAULT_FFMPEG_TIMEOUT == 1800
|
|
|
|
def test_timeout_expired_is_raised(self):
|
|
"""超时未完成时 TimeoutExpired 异常被传播。"""
|
|
with patch("video_processing.ffmpeg_utils.subprocess.run") as mock_run:
|
|
mock_run.side_effect = subprocess.TimeoutExpired(cmd=["ffmpeg", "test"], timeout=1)
|
|
with pytest.raises(subprocess.TimeoutExpired):
|
|
run_ffmpeg(["ffmpeg", "test"])
|
|
|
|
def test_custom_timeout(self):
|
|
"""支持自定义超时时间。"""
|
|
with patch("video_processing.ffmpeg_utils.subprocess.run") as mock_run:
|
|
mock_run.side_effect = subprocess.TimeoutExpired(cmd=["ffmpeg"], timeout=5)
|
|
with pytest.raises(subprocess.TimeoutExpired):
|
|
run_ffmpeg(["ffmpeg", "test"], timeout=5)
|
|
|
|
def test_none_timeout_disables_protection(self):
|
|
"""timeout=None 可以禁用超时保护（不推荐）。"""
|
|
with patch("video_processing.ffmpeg_utils.subprocess.run") as mock_run:
|
|
mock_result = MagicMock()
|
|
mock_result.stdout = ""
|
|
mock_result.stderr = ""
|
|
mock_run.return_value = mock_result
|
|
run_ffmpeg(["ffmpeg", "test"], timeout=None)
|
|
# 验证 timeout=None 被传递
|
|
call_kwargs = mock_run.call_args.kwargs
|
|
assert call_kwargs["timeout"] is None
|
|
|
|
def test_called_process_error_still_raised(self):
|
|
"""超时异常不影响原有 CalledProcessError 的抛出。"""
|
|
with patch("video_processing.ffmpeg_utils.subprocess.run") as mock_run:
|
|
mock_run.side_effect = subprocess.CalledProcessError(returncode=1, cmd=["ffmpeg"], stderr="error msg")
|
|
with pytest.raises(subprocess.CalledProcessError):
|
|
run_ffmpeg(["ffmpeg", "test"])
|
|
|
|
|
|
# ── probe_video_info 超时保护 ────────────────────────────────────────────────
|
|
|
|
|
|
class TestProbeVideoInfoTimeout:
|
|
"""probe_video_info 超时保护测试。"""
|
|
|
|
def test_probe_uses_timeout(self):
|
|
"""probe_video_info 调用 ffprobe 时应设置 timeout=15。"""
|
|
with patch("video_processing.ffmpeg_utils.subprocess.run") as mock_run:
|
|
mock_run.side_effect = subprocess.TimeoutExpired(cmd=["ffprobe"], timeout=15)
|
|
# 超时异常被捕获，返回默认值
|
|
result = probe_video_info("/tmp/test.mp4")
|
|
assert result["width"] == 1280 # DEFAULT_OUTPUT_WIDTH
|
|
assert result["height"] == 720 # DEFAULT_OUTPUT_HEIGHT
|
|
|
|
def test_probe_success(self):
|
|
"""正常情况应解析 ffprobe JSON 输出。"""
|
|
fake_output = """
|
|
{
|
|
"streams": [{"width": 1920, "height": 1080, "r_frame_rate": "30/1", "duration": "10.5"}],
|
|
"format": {"duration": "10.5"}
|
|
}
|
|
"""
|
|
with patch("video_processing.ffmpeg_utils.subprocess.run") as mock_run:
|
|
mock_result = MagicMock()
|
|
mock_result.stdout = fake_output
|
|
mock_run.return_value = mock_result
|
|
result = probe_video_info("/tmp/test.mp4")
|
|
assert result["width"] == 1920
|
|
assert result["height"] == 1080
|
|
assert abs(result["duration"] - 10.5) < 0.01
|