f523548eee
CI/CD Pipeline / Check if frontend-only change (push) Has been skipped
CI/CD Pipeline / Check push changed paths (pull_request) Has been skipped
CI/CD Pipeline / PR Build API Image (push) Has been skipped
CI/CD Pipeline / PR Build Web Image (push) Has been skipped
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / PR Build Worker Image (push) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (push) Successful in 3s
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 3s
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (pull_request) Successful in 3s
CI/CD Pipeline / Validate - Style (pull_request) Has been skipped
CI/CD Pipeline / Validate - Security (pull_request) Has been skipped
CI/CD Pipeline / Check push changed paths (push) Successful in 5s
CI/CD Pipeline / Validate - Python (mypy + alembic) (pull_request) Has been skipped
CI/CD Pipeline / Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Frontend Lint (push) Has been skipped
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 12s
CI/CD Pipeline / PR Build Web Image (pull_request) Successful in 12s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 24s
CI/CD Pipeline / Build Staging Web Image (push) Successful in 21s
CI/CD Pipeline / Build Staging API Image (push) Successful in 29s
CI/CD Pipeline / Build Production API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 22s
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / CI Gate (pull_request) Successful in 4s
CI/CD Pipeline / Retag skipped Staging API Image (push) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging Web Image (push) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging Worker Image (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / Canary Release to Production (pull_request) Has been skipped
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 1m45s
CI/CD Pipeline / Validate - Python (mypy + alembic) (push) Successful in 2m0s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 1m16s
CI/CD Pipeline / Integration Tests (push) Successful in 2m18s
CI/CD Pipeline / Validate - Style (push) Successful in 2m57s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m6s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 1m38s
CI/CD Pipeline / Frontend Unit Tests (push) Failing after 5m8s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 3m10s
CI/CD Pipeline / Validate - Security (push) Successful in 5m45s
AI Code Review / AI Code Review (pull_request) Successful in 6m19s
CI/CD Pipeline / Staging E2E Tests (push) Successful in 5m12s
CI/CD Pipeline / Unit Tests (push) Successful in 8m15s
CI/CD Pipeline / Build Production API Image (push) Has been skipped
CI/CD Pipeline / Build Production Web Image (push) Has been skipped
CI/CD Pipeline / Build Production Worker Image (push) Has been skipped
CI/CD Pipeline / CI Gate (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Canary Release to Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
Co-authored-by: xiaoxia <dev@xiaoxiajianji.com> Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>
208 lines
7.8 KiB
Python
208 lines
7.8 KiB
Python
"""#1664 随机边缘裁剪降重功能测试"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, call, patch
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(ROOT / "apps" / "worker"))
|
|
sys.path.insert(0, str(ROOT / "apps" / "api"))
|
|
sys.path.insert(0, str(ROOT / "packages"))
|
|
|
|
|
|
from video_processing.ffmpeg_utils import random_edge_crop
|
|
|
|
|
|
class TestRandomEdgeCropBasic:
|
|
"""基本功能测试"""
|
|
|
|
def test_returns_input_path_when_output_none(self, tmp_path):
|
|
"""output_path=None 时覆盖原文件并返回 input_path"""
|
|
input_file = tmp_path / "input.mp4"
|
|
input_file.write_bytes(b"fake video data")
|
|
|
|
fake_info = {"width": 1920, "height": 1080, "duration": 10, "fps": 30}
|
|
with (
|
|
patch("video_processing.ffmpeg_utils.probe_video_info", return_value=fake_info),
|
|
patch("video_processing.ffmpeg_utils.run_ffmpeg"),
|
|
):
|
|
result = random_edge_crop(input_file)
|
|
|
|
assert result == input_file
|
|
|
|
def test_returns_output_path_when_specified(self, tmp_path):
|
|
"""指定 output_path 时返回该路径"""
|
|
input_file = tmp_path / "input.mp4"
|
|
input_file.write_bytes(b"fake video data")
|
|
output_file = tmp_path / "output.mp4"
|
|
|
|
fake_info = {"width": 1920, "height": 1080, "duration": 10, "fps": 30}
|
|
with (
|
|
patch("video_processing.ffmpeg_utils.probe_video_info", return_value=fake_info),
|
|
patch("video_processing.ffmpeg_utils.run_ffmpeg"),
|
|
):
|
|
result = random_edge_crop(input_file, output_file)
|
|
|
|
assert result == output_file
|
|
|
|
def test_skip_when_invalid_resolution(self, tmp_path):
|
|
"""无法获取有效分辨率时跳过裁剪"""
|
|
input_file = tmp_path / "input.mp4"
|
|
input_file.write_bytes(b"fake video data")
|
|
|
|
fake_info = {"width": 0, "height": 0, "duration": 10, "fps": 30}
|
|
with (
|
|
patch("video_processing.ffmpeg_utils.probe_video_info", return_value=fake_info),
|
|
patch("video_processing.ffmpeg_utils.run_ffmpeg") as mock_ffmpeg,
|
|
):
|
|
result = random_edge_crop(input_file)
|
|
|
|
assert result == input_file
|
|
mock_ffmpeg.assert_not_called()
|
|
|
|
|
|
class TestRandomEdgeCropFFmpeg:
|
|
"""FFmpeg 调用参数验证"""
|
|
|
|
def test_ffmpeg_crop_and_scale_filter(self, tmp_path):
|
|
"""生成的 ffmpeg 滤镜包含 crop + scale"""
|
|
input_file = tmp_path / "input.mp4"
|
|
input_file.write_bytes(b"fake video data")
|
|
|
|
# 固定随机值以便验证
|
|
fake_info = {"width": 1000, "height": 1000, "duration": 10, "fps": 30}
|
|
|
|
with (
|
|
patch("video_processing.ffmpeg_utils.probe_video_info", return_value=fake_info),
|
|
patch("video_processing.ffmpeg_utils.run_ffmpeg") as mock_ffmpeg,
|
|
patch("random.uniform", side_effect=[0.03, 0.03, 0.03, 0.03]),
|
|
):
|
|
random_edge_crop(input_file)
|
|
|
|
mock_ffmpeg.assert_called_once()
|
|
cmd = mock_ffmpeg.call_args[0][0]
|
|
# 找到 -vf 参数
|
|
vf_idx = cmd.index("-vf")
|
|
vf_value = cmd[vf_idx + 1]
|
|
assert "crop=" in vf_value
|
|
assert "scale=1000:1000" in vf_value
|
|
|
|
def test_crop_amounts_within_range(self, tmp_path):
|
|
"""裁剪量在 2%~5% 范围内"""
|
|
input_file = tmp_path / "input.mp4"
|
|
input_file.write_bytes(b"fake video data")
|
|
|
|
fake_info = {"width": 1000, "height": 1000, "duration": 10, "fps": 30}
|
|
|
|
with (
|
|
patch("video_processing.ffmpeg_utils.probe_video_info", return_value=fake_info),
|
|
patch("video_processing.ffmpeg_utils.run_ffmpeg") as mock_ffmpeg,
|
|
patch("random.uniform", side_effect=[0.02, 0.05, 0.02, 0.05]),
|
|
):
|
|
random_edge_crop(input_file)
|
|
|
|
cmd = mock_ffmpeg.call_args[0][0]
|
|
vf_idx = cmd.index("-vf")
|
|
vf_value = cmd[vf_idx + 1]
|
|
# crop_top=20, crop_bottom=50, crop_left=20, crop_right=50
|
|
# new_w = 1000-20-50 = 930, new_h = 1000-20-50 = 930
|
|
# x_offset = 20, y_offset = 20
|
|
assert "crop=930:930:20:20" in vf_value
|
|
|
|
def test_uses_libx264_codec(self, tmp_path):
|
|
"""使用 libx264 编码"""
|
|
input_file = tmp_path / "input.mp4"
|
|
input_file.write_bytes(b"fake video data")
|
|
|
|
fake_info = {"width": 1920, "height": 1080, "duration": 10, "fps": 30}
|
|
with (
|
|
patch("video_processing.ffmpeg_utils.probe_video_info", return_value=fake_info),
|
|
patch("video_processing.ffmpeg_utils.run_ffmpeg") as mock_ffmpeg,
|
|
):
|
|
random_edge_crop(input_file)
|
|
|
|
cmd = mock_ffmpeg.call_args[0][0]
|
|
assert "-c:v" in cmd
|
|
assert cmd[cmd.index("-c:v") + 1] == "libx264"
|
|
|
|
|
|
class TestRandomEdgeCropErrorHandling:
|
|
"""错误处理测试"""
|
|
|
|
def test_ffmpeg_failure_raises_exception(self, tmp_path):
|
|
"""ffmpeg 失败时抛出异常"""
|
|
input_file = tmp_path / "input.mp4"
|
|
input_file.write_bytes(b"fake video data")
|
|
|
|
fake_info = {"width": 1920, "height": 1080, "duration": 10, "fps": 30}
|
|
with (
|
|
patch("video_processing.ffmpeg_utils.probe_video_info", return_value=fake_info),
|
|
patch(
|
|
"video_processing.ffmpeg_utils.run_ffmpeg",
|
|
side_effect=subprocess.CalledProcessError(1, "ffmpeg"),
|
|
),
|
|
):
|
|
with pytest.raises(subprocess.CalledProcessError):
|
|
random_edge_crop(input_file)
|
|
|
|
def test_probe_failure_propagates(self, tmp_path):
|
|
"""probe_video_info 失败时异常传播"""
|
|
input_file = tmp_path / "input.mp4"
|
|
input_file.write_bytes(b"fake video data")
|
|
|
|
with patch(
|
|
"video_processing.ffmpeg_utils.probe_video_info",
|
|
side_effect=RuntimeError("probe failed"),
|
|
):
|
|
with pytest.raises(RuntimeError, match="probe failed"):
|
|
random_edge_crop(input_file)
|
|
|
|
|
|
class TestRandomEdgeCropEvenDimensions:
|
|
"""偶数尺寸处理测试"""
|
|
|
|
def test_odd_crop_dimensions_adjusted_to_even(self, tmp_path):
|
|
"""裁剪后尺寸为奇数时自动调整为偶数"""
|
|
input_file = tmp_path / "input.mp4"
|
|
input_file.write_bytes(b"fake video data")
|
|
|
|
# 1000 - 3 (top) - 4 (bottom) = 993 → 调整为 992
|
|
# 1000 - 3 (left) - 4 (right) = 993 → 调整为 992
|
|
fake_info = {"width": 1000, "height": 1000, "duration": 10, "fps": 30}
|
|
|
|
with (
|
|
patch("video_processing.ffmpeg_utils.probe_video_info", return_value=fake_info),
|
|
patch("video_processing.ffmpeg_utils.run_ffmpeg") as mock_ffmpeg,
|
|
# side_effect 控制 uniform 返回值
|
|
# top: 0.003*1000=3, bottom: 0.004*1000=4, left: 0.003*1000=3, right: 0.004*1000=4
|
|
):
|
|
# 使用自定义 uniform 返回特定值
|
|
def fake_uniform(low, high):
|
|
# 返回特定百分比使得裁剪后尺寸为奇数
|
|
# 我们需要 crop_top=3, crop_bottom=4, crop_left=3, crop_right=4
|
|
return 0.0035 # 近似值
|
|
|
|
# 更简单的方式:直接 mock int(H * random.uniform(...)) 的结果
|
|
# 但我们直接测试最终 crop 滤镜即可
|
|
with patch("random.uniform", side_effect=[0.021, 0.022, 0.021, 0.022]):
|
|
random_edge_crop(input_file)
|
|
|
|
cmd = mock_ffmpeg.call_args[0][0]
|
|
vf_idx = cmd.index("-vf")
|
|
vf_value = cmd[vf_idx + 1]
|
|
# 提取 crop 参数并验证都是偶数
|
|
import re
|
|
|
|
crop_match = re.search(r"crop=(\d+):(\d+)", vf_value)
|
|
assert crop_match
|
|
crop_w = int(crop_match.group(1))
|
|
crop_h = int(crop_match.group(2))
|
|
assert crop_w % 2 == 0, f"crop width {crop_w} should be even"
|
|
assert crop_h % 2 == 0, f"crop height {crop_h} should be even"
|