xiaoxia
ffd99ffeb0
CI/CD Pipeline / Validate Code Quality And Tests (push) Successful in 40s
CI/CD Pipeline / Unit Tests (push) Successful in 1m5s
CI/CD Pipeline / Integration Tests (push) Successful in 1m15s
CI/CD Pipeline / Frontend Lint (push) Successful in 1m55s
CI/CD Pipeline / Build Production Runtime Images (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (push) Failing after 31m0s
CI/CD Pipeline / Staging E2E Tests (push) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (push) Has been skipped
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, "codec_type": "video", "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
|