CI Bot
5fa915cdad
CI/CD Pipeline / Check if frontend-only change (push) Has been cancelled
CI/CD Pipeline / Validate - Code Quality (push) Has been cancelled
CI/CD Pipeline / Validate - Type Check (mypy) (push) Has been cancelled
CI/CD Pipeline / Validate - Migration (alembic) (push) Has been cancelled
CI/CD Pipeline / Unit Tests (push) Has been cancelled
CI/CD Pipeline / Integration Tests (push) Has been cancelled
CI/CD Pipeline / Frontend Lint (push) Has been cancelled
CI/CD Pipeline / Frontend Unit Tests (push) Has been cancelled
CI/CD Pipeline / PR Build API Image (push) Has been cancelled
CI/CD Pipeline / PR Build Web Image (push) Has been cancelled
CI/CD Pipeline / PR Build Worker Image (push) Has been cancelled
CI/CD Pipeline / Build Staging API Image (push) Has been cancelled
CI/CD Pipeline / Build Staging Web Image (push) Has been cancelled
CI/CD Pipeline / Build Staging Worker 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 Web Image (push) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (push) Has been cancelled
CI/CD Pipeline / Deploy Production (push) Has been cancelled
CI/CD Pipeline / Production Browser E2E (push) Has been cancelled
CI/CD Pipeline / ACR Image Cleanup (push) Has been cancelled
CI/CD Pipeline / Canary Release to Production (push) Has been cancelled
404 lines
14 KiB
Python
Executable File
404 lines
14 KiB
Python
Executable File
"""SpeedEngine 纯逻辑单测 — 配置解析 + 调速滤镜 + 时长计算.
|
||
|
||
全纯函数测试，不依赖 FFmpeg 或外部服务。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
from video_processing.speed_engine import (
|
||
MAX_SPEED,
|
||
MIN_SPEED,
|
||
SpeedConfig,
|
||
SpeedEngine,
|
||
)
|
||
|
||
# ── SpeedConfig 解析 ──────────────────────────────────────────────────────
|
||
|
||
|
||
class TestSpeedConfigParse:
|
||
"""SpeedConfig.from_dict / parse 解析测试."""
|
||
|
||
def test_none_data_returns_default(self):
|
||
"""None 输入返回默认配置."""
|
||
config = SpeedConfig.parse(None)
|
||
assert config.speed == 1.0
|
||
assert config.pitch_correct is True
|
||
|
||
def test_empty_dict_returns_default(self):
|
||
"""空 dict 返回默认配置."""
|
||
config = SpeedConfig.parse({})
|
||
assert config.speed == 1.0
|
||
assert config.pitch_correct is True
|
||
|
||
def test_valid_speed_and_pitch(self):
|
||
"""正常速度和音调配置."""
|
||
config = SpeedConfig.parse({"speed": 2.0, "pitch_correct": False})
|
||
assert config.speed == 2.0
|
||
assert config.pitch_correct is False
|
||
|
||
def test_speed_as_int(self):
|
||
"""整数 speed 自动转 float."""
|
||
config = SpeedConfig.parse({"speed": 2})
|
||
assert config.speed == 2.0
|
||
assert isinstance(config.speed, float)
|
||
|
||
def test_invalid_speed_type_falls_back(self):
|
||
"""speed 类型错误回退到默认."""
|
||
config = SpeedConfig.parse({"speed": "fast"})
|
||
assert config.speed == 1.0
|
||
|
||
def test_invalid_pitch_correct_type_falls_back(self):
|
||
"""pitch_correct 非 bool 回退到 True."""
|
||
config = SpeedConfig.parse({"pitch_correct": "yes"})
|
||
assert config.pitch_correct is True
|
||
|
||
def test_non_dict_input_falls_back(self):
|
||
"""非 dict 输入回退到默认."""
|
||
config = SpeedConfig.parse("speed=2x")
|
||
assert config.speed == 1.0
|
||
assert config.pitch_correct is True
|
||
|
||
|
||
class TestSpeedConfigClamp:
|
||
"""SpeedConfig.clamp 钳制测试."""
|
||
|
||
def test_zero_speed_clamps_to_default(self):
|
||
"""speed=0 钳制到默认 1.0."""
|
||
config = SpeedConfig(speed=0.0)
|
||
config.clamp()
|
||
assert config.speed == 1.0
|
||
|
||
def test_negative_speed_clamps_to_default(self):
|
||
"""负速度钳制到默认 1.0."""
|
||
config = SpeedConfig(speed=-1.0)
|
||
config.clamp()
|
||
assert config.speed == 1.0
|
||
|
||
def test_below_min_clamps_to_min(self):
|
||
"""低于最小速度钳制到 MIN_SPEED."""
|
||
config = SpeedConfig(speed=0.1)
|
||
config.clamp()
|
||
assert config.speed == MIN_SPEED
|
||
|
||
def test_at_min_stays(self):
|
||
"""恰好在最小值保持不变."""
|
||
config = SpeedConfig(speed=MIN_SPEED)
|
||
config.clamp()
|
||
assert config.speed == MIN_SPEED
|
||
|
||
def test_above_max_clamps_to_max(self):
|
||
"""超过最大速度钳制到 MAX_SPEED."""
|
||
config = SpeedConfig(speed=5.0)
|
||
config.clamp()
|
||
assert config.speed == MAX_SPEED
|
||
|
||
def test_at_max_stays(self):
|
||
"""恰好在最大值保持不变."""
|
||
config = SpeedConfig(speed=MAX_SPEED)
|
||
config.clamp()
|
||
assert config.speed == MAX_SPEED
|
||
|
||
def test_normal_speed_stays(self):
|
||
"""正常范围内速度保持不变."""
|
||
config = SpeedConfig(speed=1.5)
|
||
config.clamp()
|
||
assert config.speed == 1.5
|
||
|
||
def test_parse_auto_clamps(self):
|
||
"""parse 自动执行 clamp."""
|
||
config = SpeedConfig.parse({"speed": 10.0})
|
||
assert config.speed == MAX_SPEED
|
||
|
||
|
||
class TestSpeedConfigIsOriginal:
|
||
"""SpeedConfig.is_original 属性测试."""
|
||
|
||
def test_default_is_original(self):
|
||
"""默认配置为原速."""
|
||
assert SpeedConfig().is_original is True
|
||
|
||
def test_exactly_one_is_original(self):
|
||
"""speed=1.0 为原速."""
|
||
assert SpeedConfig(speed=1.0).is_original is True
|
||
|
||
def test_very_close_is_original(self):
|
||
"""浮点精度接近 1.0 视为原速."""
|
||
assert SpeedConfig(speed=1.0000001).is_original is True
|
||
|
||
def test_different_speed_not_original(self):
|
||
"""非 1.0 速度不是原速."""
|
||
assert SpeedConfig(speed=2.0).is_original is False
|
||
assert SpeedConfig(speed=0.5).is_original is False
|
||
|
||
|
||
# ── SpeedEngine 滤镜构建 ──────────────────────────────────────────────────
|
||
|
||
|
||
class TestSpeedEngineBuildVideoFilter:
|
||
"""SpeedEngine.build_video_filter 视频滤镜测试."""
|
||
|
||
def setup_method(self):
|
||
self.engine = SpeedEngine()
|
||
|
||
def test_original_speed_returns_empty(self):
|
||
"""原速返回空字符串（无滤镜）."""
|
||
config = SpeedConfig(speed=1.0)
|
||
assert self.engine.build_video_filter(config) == ""
|
||
|
||
def test_double_speed(self):
|
||
"""2倍速 setpts=PTS/2."""
|
||
config = SpeedConfig(speed=2.0)
|
||
result = self.engine.build_video_filter(config)
|
||
assert result == "setpts=PTS/2.0000"
|
||
|
||
def test_half_speed(self):
|
||
"""0.5倍速 setpts=PTS/0.5."""
|
||
config = SpeedConfig(speed=0.5)
|
||
result = self.engine.build_video_filter(config)
|
||
assert result == "setpts=PTS/0.5000"
|
||
|
||
def test_quarter_speed(self):
|
||
"""0.25倍速."""
|
||
config = SpeedConfig(speed=0.25)
|
||
result = self.engine.build_video_filter(config)
|
||
assert "setpts=PTS/0.25" in result
|
||
|
||
def test_quad_speed(self):
|
||
"""4倍速."""
|
||
config = SpeedConfig(speed=4.0)
|
||
result = self.engine.build_video_filter(config)
|
||
assert "setpts=PTS/4.0" in result
|
||
|
||
def test_custom_speed_precision(self):
|
||
"""自定义速度保留4位小数."""
|
||
config = SpeedConfig(speed=1.333)
|
||
result = self.engine.build_video_filter(config)
|
||
assert result == "setpts=PTS/1.3330"
|
||
|
||
|
||
class TestSpeedEngineBuildAudioFilter:
|
||
"""SpeedEngine.build_audio_filter 音频滤镜测试."""
|
||
|
||
def setup_method(self):
|
||
self.engine = SpeedEngine()
|
||
|
||
def test_original_speed_returns_empty(self):
|
||
"""原速返回空字符串."""
|
||
config = SpeedConfig(speed=1.0)
|
||
assert self.engine.build_audio_filter(config) == ""
|
||
|
||
def test_within_single_stage_range(self):
|
||
"""单级 atempo 范围内返回一级."""
|
||
config = SpeedConfig(speed=1.5)
|
||
result = self.engine.build_audio_filter(config)
|
||
assert result == "atempo=1.5000"
|
||
|
||
def test_at_max_single_stage(self):
|
||
"""恰好 2.0 单级."""
|
||
config = SpeedConfig(speed=2.0)
|
||
result = self.engine.build_audio_filter(config)
|
||
assert result == "atempo=2.0000"
|
||
|
||
def test_at_min_single_stage(self):
|
||
"""恰好 0.5 单级."""
|
||
config = SpeedConfig(speed=0.5)
|
||
result = self.engine.build_audio_filter(config)
|
||
assert result == "atempo=0.5000"
|
||
|
||
def test_quad_speed_two_stages(self):
|
||
"""4倍速 = atempo=2.0,atempo=2.0."""
|
||
config = SpeedConfig(speed=4.0)
|
||
result = self.engine.build_audio_filter(config)
|
||
stages = result.split(",")
|
||
assert len(stages) == 2
|
||
assert stages[0] == "atempo=2.0000"
|
||
assert stages[1] == "atempo=2.0000"
|
||
|
||
def test_quarter_speed_two_stages(self):
|
||
"""0.25倍速 = atempo=0.5,atempo=0.5."""
|
||
config = SpeedConfig(speed=0.25)
|
||
result = self.engine.build_audio_filter(config)
|
||
stages = result.split(",")
|
||
assert len(stages) == 2
|
||
assert stages[0] == "atempo=0.5000"
|
||
assert stages[1] == "atempo=0.5000"
|
||
|
||
def test_triple_speed_two_stages(self):
|
||
"""3倍速 = atempo=2.0,atempo=1.5 (2.0 * 1.5 = 3.0)."""
|
||
config = SpeedConfig(speed=3.0)
|
||
result = self.engine.build_audio_filter(config)
|
||
stages = result.split(",")
|
||
assert len(stages) == 2
|
||
# 乘积应为 3.0
|
||
values = [float(s.split("=")[1]) for s in stages]
|
||
product = 1.0
|
||
for v in values:
|
||
product *= v
|
||
assert abs(product - 3.0) < 0.01
|
||
|
||
def test_low_speed_two_stages(self):
|
||
"""0.3倍速多级串联，乘积为 0.3."""
|
||
config = SpeedConfig(speed=0.3)
|
||
result = self.engine.build_audio_filter(config)
|
||
stages = result.split(",")
|
||
assert len(stages) >= 2
|
||
values = [float(s.split("=")[1]) for s in stages]
|
||
product = 1.0
|
||
for v in values:
|
||
product *= v
|
||
assert abs(product - 0.3) < 0.01
|
||
|
||
def test_all_stages_within_valid_range(self):
|
||
"""所有 atempo 级都在 [0.5, 2.0] 范围内."""
|
||
for speed in [0.25, 0.3, 0.5, 0.8, 1.5, 2.0, 3.0, 4.0]:
|
||
config = SpeedConfig(speed=speed)
|
||
result = self.engine.build_audio_filter(config)
|
||
if not result:
|
||
continue
|
||
stages = result.split(",")
|
||
for stage in stages:
|
||
val = float(stage.split("=")[1])
|
||
assert 0.5 <= val <= 2.0, f"speed={speed}, stage={val} out of range"
|
||
|
||
|
||
class TestSpeedEngineSplitAtempoStages:
|
||
"""SpeedEngine._split_atempo_stages 拆分算法测试."""
|
||
|
||
def test_single_stage_within_range(self):
|
||
"""范围内单级."""
|
||
stages = SpeedEngine._split_atempo_stages(1.5)
|
||
assert stages == [1.5]
|
||
|
||
def test_exactly_max_single_stage(self):
|
||
"""恰好 2.0 单级."""
|
||
stages = SpeedEngine._split_atempo_stages(2.0)
|
||
assert stages == [2.0]
|
||
|
||
def test_exactly_min_single_stage(self):
|
||
"""恰好 0.5 单级."""
|
||
stages = SpeedEngine._split_atempo_stages(0.5)
|
||
assert stages == [0.5]
|
||
|
||
def test_four_x_two_stages(self):
|
||
"""4.0 拆为两级 2.0."""
|
||
stages = SpeedEngine._split_atempo_stages(4.0)
|
||
assert stages == [2.0, 2.0]
|
||
|
||
def test_quarter_x_two_stages(self):
|
||
"""0.25 拆为两级 0.5."""
|
||
stages = SpeedEngine._split_atempo_stages(0.25)
|
||
assert stages == [0.5, 0.5]
|
||
|
||
def test_product_matches_original_speed(self):
|
||
"""拆分后乘积应等于原速度."""
|
||
for speed in [0.25, 0.3, 0.5, 0.8, 1.0, 1.5, 2.0, 2.5, 3.0, 4.0]:
|
||
stages = SpeedEngine._split_atempo_stages(speed)
|
||
product = 1.0
|
||
for s in stages:
|
||
product *= s
|
||
assert abs(product - speed) < 0.001, f"speed={speed}, product={product}"
|
||
|
||
|
||
class TestSpeedEngineAdjustDuration:
|
||
"""SpeedEngine.adjust_duration 时长计算测试."""
|
||
|
||
def setup_method(self):
|
||
self.engine = SpeedEngine()
|
||
|
||
def test_original_speed_same_duration(self):
|
||
"""原速时长不变."""
|
||
config = SpeedConfig(speed=1.0)
|
||
assert self.engine.adjust_duration(10.0, config) == 10.0
|
||
|
||
def test_double_speed_half_duration(self):
|
||
"""2倍速时长减半."""
|
||
config = SpeedConfig(speed=2.0)
|
||
assert self.engine.adjust_duration(10.0, config) == 5.0
|
||
|
||
def test_half_speed_double_duration(self):
|
||
"""0.5倍速时长加倍."""
|
||
config = SpeedConfig(speed=0.5)
|
||
assert self.engine.adjust_duration(10.0, config) == 20.0
|
||
|
||
def test_zero_duration_stays_zero(self):
|
||
"""零时长保持零."""
|
||
config = SpeedConfig(speed=2.0)
|
||
assert self.engine.adjust_duration(0.0, config) == 0.0
|
||
|
||
def test_negative_duration_stays(self):
|
||
"""负时长直接返回（不做调速）."""
|
||
config = SpeedConfig(speed=2.0)
|
||
assert self.engine.adjust_duration(-5.0, config) == -5.0
|
||
|
||
def test_quad_speed_quarter_duration(self):
|
||
"""4倍速时长为1/4."""
|
||
config = SpeedConfig(speed=4.0)
|
||
assert self.engine.adjust_duration(20.0, config) == 5.0
|
||
|
||
|
||
class TestSpeedEngineBuildClipSpeedFilter:
|
||
"""SpeedEngine.build_clip_speed_filter 便捷方法测试."""
|
||
|
||
def setup_method(self):
|
||
self.engine = SpeedEngine()
|
||
|
||
def test_default_speed_returns_empty_filters(self):
|
||
"""默认速度返回空滤镜."""
|
||
vf, af, config = self.engine.build_clip_speed_filter(1.0)
|
||
assert vf == ""
|
||
assert af == ""
|
||
assert config.is_original is True
|
||
|
||
def test_double_speed_filters(self):
|
||
"""2倍速返回对应滤镜."""
|
||
vf, af, config = self.engine.build_clip_speed_filter(2.0)
|
||
assert vf == "setpts=PTS/2.0000"
|
||
assert af == "atempo=2.0000"
|
||
assert config.speed == 2.0
|
||
|
||
def test_speed_gets_clamped(self):
|
||
"""超范围速度自动钳制."""
|
||
vf, af, config = self.engine.build_clip_speed_filter(10.0)
|
||
assert config.speed == MAX_SPEED
|
||
assert "setpts" in vf
|
||
|
||
def test_pitch_correct_false_still_has_audio_filter(self):
|
||
"""pitch_correct=False 也返回音频滤镜（只是方式不同，当前实现仍用atempo）."""
|
||
vf, af, config = self.engine.build_clip_speed_filter(2.0, pitch_correct=False)
|
||
assert config.pitch_correct is False
|
||
# 当前实现 pitch_correct 不影响滤镜输出（atempo 本身保持音调）
|
||
assert "atempo" in af
|
||
|
||
|
||
class TestSpeedEngineResolveClipSpeed:
|
||
"""SpeedEngine.resolve_clip_speed 速度解析测试."""
|
||
|
||
def test_no_clip_config_uses_global(self):
|
||
"""无 clip config 使用全局速度."""
|
||
assert SpeedEngine.resolve_clip_speed(None, 2.0) == 2.0
|
||
|
||
def test_empty_config_uses_global(self):
|
||
"""空 config 使用全局速度."""
|
||
assert SpeedEngine.resolve_clip_speed({}, 1.5) == 1.5
|
||
|
||
def test_zero_speed_uses_global(self):
|
||
"""playback_speed=0 使用全局."""
|
||
assert SpeedEngine.resolve_clip_speed({"playback_speed": 0}, 2.0) == 2.0
|
||
|
||
def test_valid_clip_speed(self):
|
||
"""有效 clip 速度优先."""
|
||
assert SpeedEngine.resolve_clip_speed({"playback_speed": 1.5}, 1.0) == 1.5
|
||
|
||
def test_invalid_speed_type_uses_global(self):
|
||
"""速度类型错误回退到全局."""
|
||
assert SpeedEngine.resolve_clip_speed({"playback_speed": "fast"}, 2.0) == 2.0
|
||
|
||
def test_negative_speed_uses_global(self):
|
||
"""负速度回退到全局."""
|
||
assert SpeedEngine.resolve_clip_speed({"playback_speed": -1}, 1.0) == 1.0
|
||
|
||
def test_default_global_is_one(self):
|
||
"""默认全局速度为 1.0."""
|
||
assert SpeedEngine.resolve_clip_speed({}) == 1.0
|