test(P3-1): 第55波 worker层引擎配置验证单测(+120) #847
Executable
+234
@@ -0,0 +1,234 @@
|
||||
"""
|
||||
视频拼接引擎配置与纯逻辑测试.
|
||||
|
||||
覆盖 ConcatSegment.from_dict / ConcatConfig.from_config_dict / has_effect / total_segments 等纯逻辑.
|
||||
引擎核心 render 方法依赖 FFmpeg,由集成测试覆盖.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from video_processing.concat_engine import ConcatConfig, ConcatSegment
|
||||
|
||||
|
||||
class TestConcatSegmentFromDict:
|
||||
"""ConcatSegment.from_dict 构造逻辑."""
|
||||
|
||||
def test_basic(self):
|
||||
seg = ConcatSegment.from_dict({"video_path": "/tmp/a.mp4"})
|
||||
assert seg.video_path == "/tmp/a.mp4"
|
||||
assert seg.start_time == 0.0
|
||||
assert seg.duration == 0.0
|
||||
assert seg.has_audio is True
|
||||
|
||||
def test_full_fields(self):
|
||||
seg = ConcatSegment.from_dict({
|
||||
"video_path": "/tmp/b.mp4",
|
||||
"start_time": 5.5,
|
||||
"duration": 10.0,
|
||||
"has_audio": False,
|
||||
})
|
||||
assert seg.video_path == "/tmp/b.mp4"
|
||||
assert seg.start_time == 5.5
|
||||
assert seg.duration == 10.0
|
||||
assert seg.has_audio is False
|
||||
|
||||
def test_negative_start_time_clamped(self):
|
||||
seg = ConcatSegment.from_dict({
|
||||
"video_path": "/tmp/a.mp4",
|
||||
"start_time": -1.0,
|
||||
})
|
||||
assert seg.start_time == 0.0
|
||||
|
||||
def test_negative_duration_clamped(self):
|
||||
seg = ConcatSegment.from_dict({
|
||||
"video_path": "/tmp/a.mp4",
|
||||
"duration": -5.0,
|
||||
})
|
||||
assert seg.duration == 0.0
|
||||
|
||||
def test_invalid_start_time_type_falls_back(self):
|
||||
seg = ConcatSegment.from_dict({
|
||||
"video_path": "/tmp/a.mp4",
|
||||
"start_time": "not_a_number",
|
||||
})
|
||||
assert seg.start_time == 0.0
|
||||
|
||||
def test_invalid_duration_type_falls_back(self):
|
||||
seg = ConcatSegment.from_dict({
|
||||
"video_path": "/tmp/a.mp4",
|
||||
"duration": "abc",
|
||||
})
|
||||
assert seg.duration == 0.0
|
||||
|
||||
def test_start_time_none_falls_back(self):
|
||||
seg = ConcatSegment.from_dict({
|
||||
"video_path": "/tmp/a.mp4",
|
||||
"start_time": None,
|
||||
})
|
||||
assert seg.start_time == 0.0
|
||||
|
||||
def test_empty_video_path_stored(self):
|
||||
seg = ConcatSegment.from_dict({"video_path": ""})
|
||||
assert seg.video_path == ""
|
||||
|
||||
|
||||
class TestConcatConfigFromConfigDict:
|
||||
"""ConcatConfig.from_config_dict 构造逻辑."""
|
||||
|
||||
def test_none_returns_default(self):
|
||||
cfg = ConcatConfig.from_config_dict(None)
|
||||
assert cfg.segments == []
|
||||
assert cfg.output_width == 0
|
||||
assert cfg.output_height == 0
|
||||
assert cfg.output_fps == 0.0
|
||||
assert cfg.force_reencode is False
|
||||
|
||||
def test_empty_dict_returns_default(self):
|
||||
cfg = ConcatConfig.from_config_dict({})
|
||||
assert cfg.segments == []
|
||||
|
||||
def test_non_dict_returns_default(self):
|
||||
cfg = ConcatConfig.from_config_dict("not a dict")
|
||||
assert cfg.segments == []
|
||||
|
||||
def test_single_segment(self):
|
||||
cfg = ConcatConfig.from_config_dict({
|
||||
"segments": [
|
||||
{"video_path": "/tmp/a.mp4", "duration": 5.0},
|
||||
],
|
||||
})
|
||||
assert len(cfg.segments) == 1
|
||||
assert cfg.segments[0].video_path == "/tmp/a.mp4"
|
||||
assert cfg.segments[0].duration == 5.0
|
||||
|
||||
def test_multiple_segments(self):
|
||||
cfg = ConcatConfig.from_config_dict({
|
||||
"segments": [
|
||||
{"video_path": "/tmp/a.mp4"},
|
||||
{"video_path": "/tmp/b.mp4", "start_time": 2.0},
|
||||
{"video_path": "/tmp/c.mp4", "duration": 3.0, "has_audio": False},
|
||||
],
|
||||
})
|
||||
assert len(cfg.segments) == 3
|
||||
assert cfg.segments[0].video_path == "/tmp/a.mp4"
|
||||
assert cfg.segments[1].start_time == 2.0
|
||||
assert cfg.segments[2].has_audio is False
|
||||
|
||||
def test_invalid_segments_filtered(self):
|
||||
cfg = ConcatConfig.from_config_dict({
|
||||
"segments": [
|
||||
{"video_path": "/tmp/valid.mp4"},
|
||||
{"video_path": ""}, # 空路径被过滤
|
||||
{"not_video_path": "xxx"}, # 没有video_path被过滤
|
||||
"not_a_dict", # 不是dict被过滤
|
||||
None, # None被过滤
|
||||
],
|
||||
})
|
||||
assert len(cfg.segments) == 1
|
||||
assert cfg.segments[0].video_path == "/tmp/valid.mp4"
|
||||
|
||||
def test_segments_not_a_list(self):
|
||||
cfg = ConcatConfig.from_config_dict({
|
||||
"segments": "not_a_list",
|
||||
})
|
||||
assert cfg.segments == []
|
||||
|
||||
def test_output_params(self):
|
||||
cfg = ConcatConfig.from_config_dict({
|
||||
"segments": [],
|
||||
"output_width": 1920,
|
||||
"output_height": 1080,
|
||||
"output_fps": 30.0,
|
||||
"force_reencode": True,
|
||||
})
|
||||
assert cfg.output_width == 1920
|
||||
assert cfg.output_height == 1080
|
||||
assert cfg.output_fps == 30.0
|
||||
assert cfg.force_reencode is True
|
||||
|
||||
def test_negative_output_params_clamped(self):
|
||||
cfg = ConcatConfig.from_config_dict({
|
||||
"segments": [],
|
||||
"output_width": -100,
|
||||
"output_height": -50,
|
||||
"output_fps": -1.0,
|
||||
})
|
||||
assert cfg.output_width == 0
|
||||
assert cfg.output_height == 0
|
||||
assert cfg.output_fps == 0.0
|
||||
|
||||
def test_invalid_output_params_fall_back(self):
|
||||
cfg = ConcatConfig.from_config_dict({
|
||||
"segments": [],
|
||||
"output_width": "abc",
|
||||
"output_height": None,
|
||||
"output_fps": "xyz",
|
||||
})
|
||||
assert cfg.output_width == 0
|
||||
assert cfg.output_height == 0
|
||||
assert cfg.output_fps == 0.0
|
||||
|
||||
def test_transition_config(self):
|
||||
cfg = ConcatConfig.from_config_dict({
|
||||
"segments": [],
|
||||
"transition": "crossfade",
|
||||
"transition_duration": 1.0,
|
||||
})
|
||||
assert cfg.transition == "crossfade"
|
||||
assert cfg.transition_duration == 1.0
|
||||
|
||||
def test_transition_duration_minimum(self):
|
||||
"""transition_duration 不能小于 0.1."""
|
||||
cfg = ConcatConfig.from_config_dict({
|
||||
"segments": [],
|
||||
"transition_duration": 0.01,
|
||||
})
|
||||
assert cfg.transition_duration >= 0.1
|
||||
|
||||
def test_default_values(self):
|
||||
cfg = ConcatConfig.from_config_dict({"segments": []})
|
||||
assert cfg.transition == "none"
|
||||
assert cfg.transition_duration == 0.3
|
||||
assert cfg.force_reencode is False
|
||||
|
||||
|
||||
class TestConcatConfigProperties:
|
||||
"""has_effect / total_segments 属性."""
|
||||
|
||||
def test_has_effect_two_or_more_valid(self):
|
||||
cfg = ConcatConfig(segments=[
|
||||
ConcatSegment(video_path="/tmp/a.mp4"),
|
||||
ConcatSegment(video_path="/tmp/b.mp4"),
|
||||
])
|
||||
assert cfg.has_effect is True
|
||||
|
||||
def test_no_effect_one_segment(self):
|
||||
cfg = ConcatConfig(segments=[
|
||||
ConcatSegment(video_path="/tmp/a.mp4"),
|
||||
])
|
||||
assert cfg.has_effect is False
|
||||
|
||||
def test_no_effect_zero_segments(self):
|
||||
cfg = ConcatConfig(segments=[])
|
||||
assert cfg.has_effect is False
|
||||
|
||||
def test_no_effect_empty_paths(self):
|
||||
cfg = ConcatConfig(segments=[
|
||||
ConcatSegment(video_path=""),
|
||||
ConcatSegment(video_path=""),
|
||||
])
|
||||
assert cfg.has_effect is False
|
||||
|
||||
def test_total_segments(self):
|
||||
cfg = ConcatConfig(segments=[
|
||||
ConcatSegment(video_path="/tmp/a.mp4"),
|
||||
ConcatSegment(video_path=""),
|
||||
ConcatSegment(video_path="/tmp/b.mp4"),
|
||||
])
|
||||
assert cfg.total_segments == 2
|
||||
|
||||
def test_total_segments_empty(self):
|
||||
cfg = ConcatConfig(segments=[])
|
||||
assert cfg.total_segments == 0
|
||||
Executable
+302
@@ -0,0 +1,302 @@
|
||||
"""
|
||||
片头片尾引擎配置与纯逻辑测试.
|
||||
|
||||
覆盖 IntroOutroConfig.from_dict / validate / has_intro / has_outro 等纯逻辑.
|
||||
引擎核心 render 方法依赖 FFmpeg,由集成测试覆盖.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
from video_processing.intro_outro_engine import IntroOutroConfig
|
||||
|
||||
|
||||
class TestIntroOutroConfigFromDict:
|
||||
"""from_dict 构造逻辑."""
|
||||
|
||||
def test_none_returns_default_disabled(self):
|
||||
cfg = IntroOutroConfig.from_dict(None)
|
||||
assert cfg.enabled is False
|
||||
assert cfg.intro_type == "none"
|
||||
assert cfg.outro_type == "none"
|
||||
|
||||
def test_empty_dict_returns_default_disabled(self):
|
||||
cfg = IntroOutroConfig.from_dict({})
|
||||
assert cfg.enabled is False
|
||||
|
||||
def test_enabled_false_returns_default_disabled(self):
|
||||
cfg = IntroOutroConfig.from_dict({"enabled": False})
|
||||
assert cfg.enabled is False
|
||||
|
||||
def test_enabled_with_video_intro(self):
|
||||
cfg = IntroOutroConfig.from_dict({
|
||||
"enabled": True,
|
||||
"intro": {
|
||||
"type": "video",
|
||||
"video_path": "/tmp/intro.mp4",
|
||||
"duration": 5.0,
|
||||
},
|
||||
"outro": {"type": "none"},
|
||||
})
|
||||
assert cfg.enabled is True
|
||||
assert cfg.intro_type == "video"
|
||||
assert cfg.intro_video_path == "/tmp/intro.mp4"
|
||||
assert cfg.intro_duration == 5.0
|
||||
|
||||
def test_enabled_with_text_intro(self):
|
||||
cfg = IntroOutroConfig.from_dict({
|
||||
"enabled": True,
|
||||
"intro": {
|
||||
"type": "text",
|
||||
"title": "Hello",
|
||||
"subtitle": "World",
|
||||
"background": "#ffffff",
|
||||
"title_color": "black",
|
||||
"title_size": 64,
|
||||
"duration": 2.5,
|
||||
},
|
||||
"outro": {"type": "none"},
|
||||
})
|
||||
assert cfg.enabled is True
|
||||
assert cfg.intro_type == "text"
|
||||
assert cfg.intro_title == "Hello"
|
||||
assert cfg.intro_subtitle == "World"
|
||||
assert cfg.intro_background == "#ffffff"
|
||||
assert cfg.intro_title_color == "black"
|
||||
assert cfg.intro_title_size == 64
|
||||
assert cfg.intro_duration == 2.5
|
||||
|
||||
def test_enabled_with_video_outro(self):
|
||||
cfg = IntroOutroConfig.from_dict({
|
||||
"enabled": True,
|
||||
"intro": {"type": "none"},
|
||||
"outro": {
|
||||
"type": "video",
|
||||
"video_path": "/tmp/outro.mp4",
|
||||
"duration": 4.0,
|
||||
},
|
||||
})
|
||||
assert cfg.enabled is True
|
||||
assert cfg.outro_type == "video"
|
||||
assert cfg.outro_video_path == "/tmp/outro.mp4"
|
||||
assert cfg.outro_duration == 4.0
|
||||
|
||||
def test_enabled_with_text_outro_default_values(self):
|
||||
cfg = IntroOutroConfig.from_dict({
|
||||
"enabled": True,
|
||||
"intro": {"type": "none"},
|
||||
"outro": {"type": "text"},
|
||||
})
|
||||
assert cfg.outro_title == "感谢观看"
|
||||
assert cfg.outro_subtitle == "点赞关注不迷路"
|
||||
assert cfg.outro_title_size == 48
|
||||
assert cfg.outro_duration == 3.0
|
||||
|
||||
def test_video_key_fallback(self):
|
||||
"""video 字段作为 video_path 的 fallback."""
|
||||
cfg = IntroOutroConfig.from_dict({
|
||||
"enabled": True,
|
||||
"intro": {
|
||||
"type": "video",
|
||||
"video": "/tmp/fallback.mp4",
|
||||
},
|
||||
"outro": {"type": "none"},
|
||||
})
|
||||
assert cfg.intro_video_path == "/tmp/fallback.mp4"
|
||||
|
||||
def test_transition_config(self):
|
||||
cfg = IntroOutroConfig.from_dict({
|
||||
"enabled": True,
|
||||
"intro": {"type": "none"},
|
||||
"outro": {"type": "none"},
|
||||
"transition": "fade",
|
||||
"transition_duration": 1.0,
|
||||
})
|
||||
assert cfg.transition_effect == "fade"
|
||||
assert cfg.transition_duration == 1.0
|
||||
|
||||
def test_default_transition(self):
|
||||
cfg = IntroOutroConfig.from_dict({
|
||||
"enabled": True,
|
||||
"intro": {"type": "none"},
|
||||
"outro": {"type": "none"},
|
||||
})
|
||||
assert cfg.transition_effect == "fade"
|
||||
assert cfg.transition_duration == 0.5
|
||||
|
||||
|
||||
class TestIntroOutroConfigProperties:
|
||||
"""has_intro / has_outro 属性."""
|
||||
|
||||
def test_has_intro_video_type(self):
|
||||
cfg = IntroOutroConfig(
|
||||
enabled=True,
|
||||
intro_type="video",
|
||||
intro_video_path="/tmp/a.mp4",
|
||||
)
|
||||
assert cfg.has_intro is True
|
||||
|
||||
def test_has_intro_text_type(self):
|
||||
cfg = IntroOutroConfig(
|
||||
enabled=True,
|
||||
intro_type="text",
|
||||
intro_title="Hi",
|
||||
)
|
||||
assert cfg.has_intro is True
|
||||
|
||||
def test_no_intro_when_disabled(self):
|
||||
cfg = IntroOutroConfig(
|
||||
enabled=False,
|
||||
intro_type="video",
|
||||
intro_video_path="/tmp/a.mp4",
|
||||
)
|
||||
assert cfg.has_intro is False
|
||||
|
||||
def test_no_intro_when_none_type(self):
|
||||
cfg = IntroOutroConfig(
|
||||
enabled=True,
|
||||
intro_type="none",
|
||||
)
|
||||
assert cfg.has_intro is False
|
||||
|
||||
def test_has_outro_video_type(self):
|
||||
cfg = IntroOutroConfig(
|
||||
enabled=True,
|
||||
outro_type="video",
|
||||
outro_video_path="/tmp/a.mp4",
|
||||
)
|
||||
assert cfg.has_outro is True
|
||||
|
||||
def test_has_outro_text_type(self):
|
||||
cfg = IntroOutroConfig(
|
||||
enabled=True,
|
||||
outro_type="text",
|
||||
outro_title="Bye",
|
||||
)
|
||||
assert cfg.has_outro is True
|
||||
|
||||
def test_has_outro_follow_type(self):
|
||||
cfg = IntroOutroConfig(
|
||||
enabled=True,
|
||||
outro_type="follow",
|
||||
outro_title="Follow me",
|
||||
)
|
||||
assert cfg.has_outro is True
|
||||
|
||||
def test_no_outro_when_disabled(self):
|
||||
cfg = IntroOutroConfig(
|
||||
enabled=False,
|
||||
outro_type="text",
|
||||
outro_title="Bye",
|
||||
)
|
||||
assert cfg.has_outro is False
|
||||
|
||||
def test_no_outro_when_none_type(self):
|
||||
cfg = IntroOutroConfig(
|
||||
enabled=True,
|
||||
outro_type="none",
|
||||
)
|
||||
assert cfg.has_outro is False
|
||||
|
||||
|
||||
class TestIntroOutroConfigValidate:
|
||||
"""validate 校验逻辑."""
|
||||
|
||||
def test_disabled_is_valid(self):
|
||||
cfg = IntroOutroConfig(enabled=False)
|
||||
ok, msg = cfg.validate()
|
||||
assert ok is True
|
||||
assert msg == ""
|
||||
|
||||
def test_video_intro_missing_path(self):
|
||||
cfg = IntroOutroConfig(
|
||||
enabled=True,
|
||||
intro_type="video",
|
||||
intro_video_path="",
|
||||
outro_type="none",
|
||||
)
|
||||
ok, msg = cfg.validate()
|
||||
assert ok is False
|
||||
assert "video_path" in msg
|
||||
|
||||
def test_text_intro_missing_title(self):
|
||||
cfg = IntroOutroConfig(
|
||||
enabled=True,
|
||||
intro_type="text",
|
||||
intro_title="",
|
||||
outro_type="none",
|
||||
)
|
||||
ok, msg = cfg.validate()
|
||||
assert ok is False
|
||||
assert "title" in msg
|
||||
|
||||
def test_video_outro_missing_path(self):
|
||||
cfg = IntroOutroConfig(
|
||||
enabled=True,
|
||||
intro_type="none",
|
||||
outro_type="video",
|
||||
outro_video_path="",
|
||||
)
|
||||
ok, msg = cfg.validate()
|
||||
assert ok is False
|
||||
assert "video_path" in msg
|
||||
|
||||
def test_text_outro_missing_title(self):
|
||||
cfg = IntroOutroConfig(
|
||||
enabled=True,
|
||||
intro_type="none",
|
||||
outro_type="text",
|
||||
outro_title="",
|
||||
)
|
||||
ok, msg = cfg.validate()
|
||||
assert ok is False
|
||||
assert "title" in msg
|
||||
|
||||
def test_intro_duration_zero(self):
|
||||
cfg = IntroOutroConfig(
|
||||
enabled=True,
|
||||
intro_type="text",
|
||||
intro_title="Hi",
|
||||
intro_duration=0,
|
||||
outro_type="none",
|
||||
)
|
||||
ok, msg = cfg.validate()
|
||||
assert ok is False
|
||||
assert "片头时长" in msg
|
||||
|
||||
def test_intro_duration_negative(self):
|
||||
cfg = IntroOutroConfig(
|
||||
enabled=True,
|
||||
intro_type="text",
|
||||
intro_title="Hi",
|
||||
intro_duration=-1.0,
|
||||
outro_type="none",
|
||||
)
|
||||
ok, msg = cfg.validate()
|
||||
assert ok is False
|
||||
assert "片头时长" in msg
|
||||
|
||||
def test_outro_duration_zero(self):
|
||||
cfg = IntroOutroConfig(
|
||||
enabled=True,
|
||||
intro_type="none",
|
||||
outro_type="text",
|
||||
outro_title="Bye",
|
||||
outro_duration=0,
|
||||
)
|
||||
ok, msg = cfg.validate()
|
||||
assert ok is False
|
||||
assert "片尾时长" in msg
|
||||
|
||||
def test_valid_full_config(self):
|
||||
cfg = IntroOutroConfig(
|
||||
enabled=True,
|
||||
intro_type="video",
|
||||
intro_video_path="/tmp/intro.mp4",
|
||||
intro_duration=3.0,
|
||||
outro_type="text",
|
||||
outro_title="Thanks",
|
||||
outro_duration=2.0,
|
||||
)
|
||||
ok, msg = cfg.validate()
|
||||
assert ok is True
|
||||
assert msg == ""
|
||||
Executable
+346
@@ -0,0 +1,346 @@
|
||||
"""
|
||||
多轨道混音引擎配置与纯逻辑测试.
|
||||
|
||||
覆盖 AudioTrack.from_dict / MultiTrackMixConfig.from_config_dict / has_effect 等纯逻辑.
|
||||
引擎核心混音方法依赖 FFmpeg,由集成测试覆盖.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from video_processing.multi_track_mixer import (
|
||||
DEFAULT_VOLUMES,
|
||||
MAX_AUDIO_TRACKS,
|
||||
TRACK_TYPE_AMBIENT,
|
||||
TRACK_TYPE_BGM,
|
||||
TRACK_TYPE_MAIN,
|
||||
TRACK_TYPE_SFX,
|
||||
TRACK_TYPE_VOICEOVER,
|
||||
AudioTrack,
|
||||
MultiTrackMixConfig,
|
||||
)
|
||||
|
||||
|
||||
class TestTrackConstants:
|
||||
"""轨道类型常量与默认值."""
|
||||
|
||||
def test_track_types_exist(self):
|
||||
assert TRACK_TYPE_MAIN == "main"
|
||||
assert TRACK_TYPE_BGM == "bgm"
|
||||
assert TRACK_TYPE_VOICEOVER == "voiceover"
|
||||
assert TRACK_TYPE_SFX == "sfx"
|
||||
assert TRACK_TYPE_AMBIENT == "ambient"
|
||||
|
||||
def test_max_tracks(self):
|
||||
assert MAX_AUDIO_TRACKS == 8
|
||||
|
||||
def test_default_volumes(self):
|
||||
assert DEFAULT_VOLUMES[TRACK_TYPE_MAIN] == 1.0
|
||||
assert DEFAULT_VOLUMES[TRACK_TYPE_BGM] == 0.3
|
||||
assert DEFAULT_VOLUMES[TRACK_TYPE_VOICEOVER] == 1.0
|
||||
assert DEFAULT_VOLUMES[TRACK_TYPE_SFX] == 0.7
|
||||
assert DEFAULT_VOLUMES[TRACK_TYPE_AMBIENT] == 0.2
|
||||
|
||||
|
||||
class TestAudioTrackFromDict:
|
||||
"""AudioTrack.from_dict 构造逻辑."""
|
||||
|
||||
def test_basic(self):
|
||||
track = AudioTrack.from_dict({
|
||||
"track_id": "t1",
|
||||
"track_type": "bgm",
|
||||
"audio_path": "/tmp/bgm.mp3",
|
||||
})
|
||||
assert track.track_id == "t1"
|
||||
assert track.track_type == "bgm"
|
||||
assert track.audio_path == "/tmp/bgm.mp3"
|
||||
assert track.volume == 0.3 # bgm 默认音量
|
||||
|
||||
def test_custom_volume(self):
|
||||
track = AudioTrack.from_dict({
|
||||
"track_id": "t1",
|
||||
"track_type": "main",
|
||||
"audio_path": "/tmp/main.wav",
|
||||
"volume": 0.8,
|
||||
})
|
||||
assert track.volume == 0.8
|
||||
|
||||
def test_volume_clamped_to_zero(self):
|
||||
track = AudioTrack.from_dict({
|
||||
"track_id": "t1",
|
||||
"track_type": "sfx",
|
||||
"audio_path": "/tmp/sfx.wav",
|
||||
"volume": -1.0,
|
||||
})
|
||||
assert track.volume == 0.0
|
||||
|
||||
def test_volume_clamped_to_max(self):
|
||||
track = AudioTrack.from_dict({
|
||||
"track_id": "t1",
|
||||
"track_type": "sfx",
|
||||
"audio_path": "/tmp/sfx.wav",
|
||||
"volume": 3.0,
|
||||
})
|
||||
assert track.volume == 2.0
|
||||
|
||||
def test_invalid_volume_falls_back_to_default(self):
|
||||
track = AudioTrack.from_dict({
|
||||
"track_id": "t1",
|
||||
"track_type": "bgm",
|
||||
"audio_path": "/tmp/bgm.mp3",
|
||||
"volume": "not_a_number",
|
||||
})
|
||||
assert track.volume == 0.3 # bgm 默认
|
||||
|
||||
def test_none_volume_falls_back(self):
|
||||
track = AudioTrack.from_dict({
|
||||
"track_id": "t1",
|
||||
"track_type": "voiceover",
|
||||
"audio_path": "/tmp/vo.wav",
|
||||
"volume": None,
|
||||
})
|
||||
assert track.volume == 1.0 # voiceover 默认
|
||||
|
||||
def test_unknown_track_type_default_volume(self):
|
||||
track = AudioTrack.from_dict({
|
||||
"track_id": "t1",
|
||||
"track_type": "unknown_type",
|
||||
"audio_path": "/tmp/a.wav",
|
||||
})
|
||||
assert track.volume == 1.0 # 未知类型默认 1.0
|
||||
|
||||
def test_fade_in_fade_out(self):
|
||||
track = AudioTrack.from_dict({
|
||||
"track_id": "t1",
|
||||
"track_type": "bgm",
|
||||
"audio_path": "/tmp/bgm.mp3",
|
||||
"fade_in": 1.5,
|
||||
"fade_out": 2.0,
|
||||
})
|
||||
assert track.fade_in == 1.5
|
||||
assert track.fade_out == 2.0
|
||||
|
||||
def test_negative_fade_clamped(self):
|
||||
track = AudioTrack.from_dict({
|
||||
"track_id": "t1",
|
||||
"track_type": "bgm",
|
||||
"audio_path": "/tmp/bgm.mp3",
|
||||
"fade_in": -0.5,
|
||||
"fade_out": -1.0,
|
||||
})
|
||||
assert track.fade_in == 0.0
|
||||
assert track.fade_out == 0.0
|
||||
|
||||
def test_invalid_fade_falls_back(self):
|
||||
track = AudioTrack.from_dict({
|
||||
"track_id": "t1",
|
||||
"track_type": "bgm",
|
||||
"audio_path": "/tmp/bgm.mp3",
|
||||
"fade_in": "abc",
|
||||
"fade_out": None,
|
||||
})
|
||||
assert track.fade_in == 0.0
|
||||
assert track.fade_out == 0.0
|
||||
|
||||
def test_start_time_and_duration(self):
|
||||
track = AudioTrack.from_dict({
|
||||
"track_id": "t1",
|
||||
"track_type": "sfx",
|
||||
"audio_path": "/tmp/sfx.wav",
|
||||
"start_time": 5.0,
|
||||
"duration": 3.0,
|
||||
})
|
||||
assert track.start_time == 5.0
|
||||
assert track.duration == 3.0
|
||||
|
||||
def test_negative_start_time_clamped(self):
|
||||
track = AudioTrack.from_dict({
|
||||
"track_id": "t1",
|
||||
"track_type": "bgm",
|
||||
"audio_path": "/tmp/bgm.mp3",
|
||||
"start_time": -10.0,
|
||||
"duration": -2.0,
|
||||
})
|
||||
assert track.start_time == 0.0
|
||||
assert track.duration == 0.0
|
||||
|
||||
def test_invalid_time_values_fall_back(self):
|
||||
track = AudioTrack.from_dict({
|
||||
"track_id": "t1",
|
||||
"track_type": "bgm",
|
||||
"audio_path": "/tmp/bgm.mp3",
|
||||
"start_time": "invalid",
|
||||
"duration": "bad",
|
||||
})
|
||||
assert track.start_time == 0.0
|
||||
assert track.duration == 0.0
|
||||
|
||||
def test_enabled_default_true(self):
|
||||
track = AudioTrack.from_dict({
|
||||
"track_id": "t1",
|
||||
"track_type": "bgm",
|
||||
"audio_path": "/tmp/bgm.mp3",
|
||||
})
|
||||
assert track.enabled is True
|
||||
|
||||
def test_enabled_can_be_false(self):
|
||||
track = AudioTrack.from_dict({
|
||||
"track_id": "t1",
|
||||
"track_type": "bgm",
|
||||
"audio_path": "/tmp/bgm.mp3",
|
||||
"enabled": False,
|
||||
})
|
||||
assert track.enabled is False
|
||||
|
||||
|
||||
class TestMultiTrackMixConfigFromDict:
|
||||
"""MultiTrackMixConfig.from_config_dict 构造逻辑."""
|
||||
|
||||
def test_none_returns_default(self):
|
||||
cfg = MultiTrackMixConfig.from_config_dict(None)
|
||||
assert cfg.tracks == []
|
||||
assert cfg.master_volume == 1.0
|
||||
assert cfg.normalize is True
|
||||
|
||||
def test_empty_dict_returns_default(self):
|
||||
cfg = MultiTrackMixConfig.from_config_dict({})
|
||||
assert cfg.tracks == []
|
||||
|
||||
def test_non_dict_returns_default(self):
|
||||
cfg = MultiTrackMixConfig.from_config_dict([])
|
||||
assert cfg.tracks == []
|
||||
|
||||
def test_single_track(self):
|
||||
cfg = MultiTrackMixConfig.from_config_dict({
|
||||
"tracks": [
|
||||
{
|
||||
"track_id": "bgm1",
|
||||
"track_type": "bgm",
|
||||
"audio_path": "/tmp/bgm.mp3",
|
||||
"volume": 0.5,
|
||||
},
|
||||
],
|
||||
})
|
||||
assert len(cfg.tracks) == 1
|
||||
assert cfg.tracks[0].track_id == "bgm1"
|
||||
assert cfg.tracks[0].volume == 0.5
|
||||
|
||||
def test_multiple_tracks(self):
|
||||
cfg = MultiTrackMixConfig.from_config_dict({
|
||||
"tracks": [
|
||||
{"track_id": "m", "track_type": "main", "audio_path": "/tmp/m.wav"},
|
||||
{"track_id": "b", "track_type": "bgm", "audio_path": "/tmp/b.mp3"},
|
||||
{"track_id": "v", "track_type": "voiceover", "audio_path": "/tmp/v.wav"},
|
||||
],
|
||||
})
|
||||
assert len(cfg.tracks) == 3
|
||||
assert cfg.tracks[0].track_type == "main"
|
||||
assert cfg.tracks[1].track_type == "bgm"
|
||||
assert cfg.tracks[2].track_type == "voiceover"
|
||||
|
||||
def test_disabled_tracks_filtered(self):
|
||||
cfg = MultiTrackMixConfig.from_config_dict({
|
||||
"tracks": [
|
||||
{"track_id": "a", "track_type": "sfx", "audio_path": "/tmp/a.wav"},
|
||||
{"track_id": "b", "track_type": "sfx", "audio_path": "/tmp/b.wav", "enabled": False},
|
||||
{"track_id": "c", "track_type": "sfx", "audio_path": "/tmp/c.wav"},
|
||||
],
|
||||
})
|
||||
assert len(cfg.tracks) == 2
|
||||
assert all(t.track_id != "b" for t in cfg.tracks)
|
||||
|
||||
def test_empty_audio_path_filtered(self):
|
||||
cfg = MultiTrackMixConfig.from_config_dict({
|
||||
"tracks": [
|
||||
{"track_id": "valid", "track_type": "sfx", "audio_path": "/tmp/a.wav"},
|
||||
{"track_id": "empty", "track_type": "sfx", "audio_path": ""},
|
||||
],
|
||||
})
|
||||
assert len(cfg.tracks) == 1
|
||||
assert cfg.tracks[0].track_id == "valid"
|
||||
|
||||
def test_invalid_tracks_skipped(self):
|
||||
cfg = MultiTrackMixConfig.from_config_dict({
|
||||
"tracks": [
|
||||
{"track_id": "ok", "track_type": "sfx", "audio_path": "/tmp/a.wav"},
|
||||
"not_a_dict",
|
||||
None,
|
||||
{"no_audio_path": "xxx"},
|
||||
],
|
||||
})
|
||||
assert len(cfg.tracks) == 1
|
||||
|
||||
def test_tracks_not_a_list(self):
|
||||
cfg = MultiTrackMixConfig.from_config_dict({
|
||||
"tracks": "not_a_list",
|
||||
})
|
||||
assert cfg.tracks == []
|
||||
|
||||
def test_master_volume(self):
|
||||
cfg = MultiTrackMixConfig.from_config_dict({
|
||||
"tracks": [],
|
||||
"master_volume": 0.8,
|
||||
})
|
||||
assert cfg.master_volume == 0.8
|
||||
|
||||
def test_master_volume_clamped(self):
|
||||
cfg = MultiTrackMixConfig.from_config_dict({
|
||||
"tracks": [],
|
||||
"master_volume": 3.0,
|
||||
})
|
||||
assert cfg.master_volume == 2.0
|
||||
|
||||
cfg2 = MultiTrackMixConfig.from_config_dict({
|
||||
"tracks": [],
|
||||
"master_volume": -1.0,
|
||||
})
|
||||
assert cfg2.master_volume == 0.0
|
||||
|
||||
def test_invalid_master_volume_falls_back(self):
|
||||
cfg = MultiTrackMixConfig.from_config_dict({
|
||||
"tracks": [],
|
||||
"master_volume": "abc",
|
||||
})
|
||||
assert cfg.master_volume == 1.0
|
||||
|
||||
def test_normalize_and_max_output(self):
|
||||
cfg = MultiTrackMixConfig.from_config_dict({
|
||||
"tracks": [],
|
||||
"normalize": False,
|
||||
"max_output_volume": 2.0,
|
||||
})
|
||||
assert cfg.normalize is False
|
||||
assert cfg.max_output_volume == 2.0
|
||||
|
||||
def test_default_values(self):
|
||||
cfg = MultiTrackMixConfig.from_config_dict({"tracks": []})
|
||||
assert cfg.master_volume == 1.0
|
||||
assert cfg.normalize is True
|
||||
assert cfg.max_output_volume == 1.5
|
||||
|
||||
|
||||
class TestMultiTrackMixConfigProperties:
|
||||
"""has_effect 属性."""
|
||||
|
||||
def test_has_effect_with_tracks(self):
|
||||
cfg = MultiTrackMixConfig(tracks=[
|
||||
AudioTrack(track_id="t1", track_type="bgm", audio_path="/tmp/a.mp3"),
|
||||
])
|
||||
assert cfg.has_effect is True
|
||||
|
||||
def test_no_effect_empty(self):
|
||||
cfg = MultiTrackMixConfig(tracks=[])
|
||||
assert cfg.has_effect is False
|
||||
|
||||
def test_no_effect_all_disabled(self):
|
||||
cfg = MultiTrackMixConfig(tracks=[
|
||||
AudioTrack(track_id="t1", track_type="bgm", audio_path="/tmp/a.mp3", enabled=False),
|
||||
])
|
||||
assert cfg.has_effect is False
|
||||
|
||||
def test_no_effect_empty_paths(self):
|
||||
cfg = MultiTrackMixConfig(tracks=[
|
||||
AudioTrack(track_id="t1", track_type="bgm", audio_path=""),
|
||||
])
|
||||
assert cfg.has_effect is False
|
||||
Executable
+258
@@ -0,0 +1,258 @@
|
||||
"""
|
||||
水印引擎配置与纯逻辑测试.
|
||||
|
||||
覆盖 WatermarkConfig.from_dict / validate / 位置枚举等纯逻辑.
|
||||
引擎核心 render 方法依赖 FFmpeg,由集成测试覆盖.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
from video_processing.watermark_engine import WATERMARK_POSITIONS, WatermarkConfig
|
||||
|
||||
|
||||
class TestWatermarkPositions:
|
||||
"""水印位置枚举."""
|
||||
|
||||
def test_nine_positions_exist(self):
|
||||
assert len(WATERMARK_POSITIONS) == 9
|
||||
assert "top_left" in WATERMARK_POSITIONS
|
||||
assert "top_center" in WATERMARK_POSITIONS
|
||||
assert "top_right" in WATERMARK_POSITIONS
|
||||
assert "center_left" in WATERMARK_POSITIONS
|
||||
assert "center" in WATERMARK_POSITIONS
|
||||
assert "center_right" in WATERMARK_POSITIONS
|
||||
assert "bottom_left" in WATERMARK_POSITIONS
|
||||
assert "bottom_center" in WATERMARK_POSITIONS
|
||||
assert "bottom_right" in WATERMARK_POSITIONS
|
||||
|
||||
def test_position_values_are_chinese_labels(self):
|
||||
for key, label in WATERMARK_POSITIONS.items():
|
||||
assert isinstance(label, str)
|
||||
assert len(label) >= 2
|
||||
|
||||
|
||||
class TestWatermarkConfigFromDict:
|
||||
"""from_dict 构造逻辑."""
|
||||
|
||||
def test_none_returns_none(self):
|
||||
assert WatermarkConfig.from_dict(None) is None
|
||||
|
||||
def test_empty_dict_returns_none(self):
|
||||
assert WatermarkConfig.from_dict({}) is None
|
||||
|
||||
def test_enabled_false_returns_none(self):
|
||||
assert WatermarkConfig.from_dict({"enabled": False}) is None
|
||||
|
||||
def test_image_mode_without_path_returns_none(self):
|
||||
result = WatermarkConfig.from_dict({
|
||||
"enabled": True,
|
||||
"mode": "image",
|
||||
})
|
||||
assert result is None
|
||||
|
||||
def test_image_mode_with_empty_path_returns_none(self):
|
||||
result = WatermarkConfig.from_dict({
|
||||
"enabled": True,
|
||||
"mode": "image",
|
||||
"image_path": "",
|
||||
})
|
||||
assert result is None
|
||||
|
||||
def test_text_mode_without_text_returns_none(self):
|
||||
result = WatermarkConfig.from_dict({
|
||||
"enabled": True,
|
||||
"mode": "text",
|
||||
})
|
||||
assert result is None
|
||||
|
||||
def test_text_mode_with_empty_text_returns_none(self):
|
||||
result = WatermarkConfig.from_dict({
|
||||
"enabled": True,
|
||||
"mode": "text",
|
||||
"text": "",
|
||||
})
|
||||
assert result is None
|
||||
|
||||
def test_image_mode_success(self):
|
||||
cfg = WatermarkConfig.from_dict({
|
||||
"enabled": True,
|
||||
"mode": "image",
|
||||
"image_path": "/tmp/logo.png",
|
||||
"scale": 0.3,
|
||||
"opacity": 0.9,
|
||||
"position": "top_left",
|
||||
"margin_x": 30,
|
||||
"margin_y": 30,
|
||||
})
|
||||
assert cfg is not None
|
||||
assert cfg.mode == "image"
|
||||
assert cfg.image_path == "/tmp/logo.png"
|
||||
assert cfg.scale == 0.3
|
||||
assert cfg.opacity == 0.9
|
||||
assert cfg.position == "top_left"
|
||||
assert cfg.margin_x == 30
|
||||
assert cfg.margin_y == 30
|
||||
|
||||
def test_image_mode_image_key_fallback(self):
|
||||
"""image 字段作为 image_path 的 fallback."""
|
||||
cfg = WatermarkConfig.from_dict({
|
||||
"enabled": True,
|
||||
"mode": "image",
|
||||
"image": "/tmp/fallback.png",
|
||||
})
|
||||
assert cfg is not None
|
||||
assert cfg.image_path == "/tmp/fallback.png"
|
||||
|
||||
def test_text_mode_success(self):
|
||||
cfg = WatermarkConfig.from_dict({
|
||||
"enabled": True,
|
||||
"mode": "text",
|
||||
"text": "hello world",
|
||||
"font_size": 32,
|
||||
"font_color": "red",
|
||||
"position": "bottom_left",
|
||||
"scroll": True,
|
||||
"scroll_speed": 100,
|
||||
})
|
||||
assert cfg is not None
|
||||
assert cfg.mode == "text"
|
||||
assert cfg.text == "hello world"
|
||||
assert cfg.font_size == 32
|
||||
assert cfg.font_color == "red"
|
||||
assert cfg.position == "bottom_left"
|
||||
assert cfg.scroll is True
|
||||
assert cfg.scroll_speed == 100
|
||||
|
||||
def test_invalid_position_falls_back_to_bottom_right(self):
|
||||
cfg = WatermarkConfig.from_dict({
|
||||
"enabled": True,
|
||||
"mode": "text",
|
||||
"text": "test",
|
||||
"position": "invalid_position",
|
||||
})
|
||||
assert cfg is not None
|
||||
assert cfg.position == "bottom_right"
|
||||
|
||||
def test_default_values_applied(self):
|
||||
cfg = WatermarkConfig.from_dict({
|
||||
"enabled": True,
|
||||
"mode": "text",
|
||||
"text": "test",
|
||||
})
|
||||
assert cfg is not None
|
||||
assert cfg.position == "bottom_right"
|
||||
assert cfg.opacity == 0.8
|
||||
assert cfg.scale == 0.2
|
||||
assert cfg.font_size == 24
|
||||
assert cfg.font_color == "white"
|
||||
assert cfg.margin_x == 20
|
||||
assert cfg.margin_y == 20
|
||||
assert cfg.scroll is False
|
||||
assert cfg.scroll_speed == 50
|
||||
|
||||
|
||||
class TestWatermarkConfigValidate:
|
||||
"""validate 校验逻辑."""
|
||||
|
||||
def test_valid_image_config(self):
|
||||
cfg = WatermarkConfig(
|
||||
mode="image",
|
||||
image_path="/tmp/logo.png",
|
||||
position="top_right",
|
||||
opacity=0.5,
|
||||
scale=0.5,
|
||||
)
|
||||
ok, msg = cfg.validate()
|
||||
assert ok is True
|
||||
assert msg == ""
|
||||
|
||||
def test_valid_text_config(self):
|
||||
cfg = WatermarkConfig(
|
||||
mode="text",
|
||||
text="hello",
|
||||
position="center",
|
||||
opacity=1.0,
|
||||
font_size=48,
|
||||
)
|
||||
ok, msg = cfg.validate()
|
||||
assert ok is True
|
||||
assert msg == ""
|
||||
|
||||
def test_invalid_position(self):
|
||||
cfg = WatermarkConfig(mode="text", text="test", position="nowhere")
|
||||
ok, msg = cfg.validate()
|
||||
assert ok is False
|
||||
assert "不支持的位置" in msg
|
||||
|
||||
def test_opacity_below_zero(self):
|
||||
cfg = WatermarkConfig(mode="text", text="test", opacity=-0.1)
|
||||
ok, msg = cfg.validate()
|
||||
assert ok is False
|
||||
assert "透明度" in msg
|
||||
|
||||
def test_opacity_above_one(self):
|
||||
cfg = WatermarkConfig(mode="text", text="test", opacity=1.5)
|
||||
ok, msg = cfg.validate()
|
||||
assert ok is False
|
||||
assert "透明度" in msg
|
||||
|
||||
def test_opacity_zero_is_valid(self):
|
||||
cfg = WatermarkConfig(mode="text", text="test", opacity=0.0)
|
||||
ok, _ = cfg.validate()
|
||||
assert ok is True
|
||||
|
||||
def test_opacity_one_is_valid(self):
|
||||
cfg = WatermarkConfig(mode="text", text="test", opacity=1.0)
|
||||
ok, _ = cfg.validate()
|
||||
assert ok is True
|
||||
|
||||
def test_image_missing_path(self):
|
||||
cfg = WatermarkConfig(mode="image", image_path="")
|
||||
ok, msg = cfg.validate()
|
||||
assert ok is False
|
||||
assert "图片路径" in msg
|
||||
|
||||
def test_image_scale_too_small(self):
|
||||
cfg = WatermarkConfig(mode="image", image_path="/tmp/a.png", scale=0.001)
|
||||
ok, msg = cfg.validate()
|
||||
assert ok is False
|
||||
assert "缩放比例" in msg
|
||||
|
||||
def test_image_scale_too_large(self):
|
||||
cfg = WatermarkConfig(mode="image", image_path="/tmp/a.png", scale=2.0)
|
||||
ok, msg = cfg.validate()
|
||||
assert ok is False
|
||||
assert "缩放比例" in msg
|
||||
|
||||
def test_image_scale_boundary_valid(self):
|
||||
cfg = WatermarkConfig(mode="image", image_path="/tmp/a.png", scale=0.01)
|
||||
ok, _ = cfg.validate()
|
||||
assert ok is True
|
||||
|
||||
cfg2 = WatermarkConfig(mode="image", image_path="/tmp/a.png", scale=1.0)
|
||||
ok2, _ = cfg2.validate()
|
||||
assert ok2 is True
|
||||
|
||||
def test_text_missing_text(self):
|
||||
cfg = WatermarkConfig(mode="text", text="")
|
||||
ok, msg = cfg.validate()
|
||||
assert ok is False
|
||||
assert "文字内容" in msg
|
||||
|
||||
def test_text_font_size_zero(self):
|
||||
cfg = WatermarkConfig(mode="text", text="test", font_size=0)
|
||||
ok, msg = cfg.validate()
|
||||
assert ok is False
|
||||
assert "字体大小" in msg
|
||||
|
||||
def test_text_font_size_negative(self):
|
||||
cfg = WatermarkConfig(mode="text", text="test", font_size=-5)
|
||||
ok, msg = cfg.validate()
|
||||
assert ok is False
|
||||
assert "字体大小" in msg
|
||||
|
||||
def test_unsupported_mode(self):
|
||||
cfg = WatermarkConfig(mode="video", text="test")
|
||||
ok, msg = cfg.validate()
|
||||
assert ok is False
|
||||
assert "不支持的水印模式" in msg
|
||||
Reference in New Issue
Block a user