test(P3-1): 第55波 worker层引擎配置验证单测(+120) #847

Merged
xiaoxia merged 1 commits from test/wave55-worker-engines-config-validation into develop 2026-07-24 21:29:37 +08:00
4 changed files with 1140 additions and 0 deletions
+234
View File
@@ -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
+302
View File
@@ -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 == ""
+346
View File
@@ -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
+258
View File
@@ -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