style: black + isort format
CI/CD Pipeline / Production Browser E2E (pull_request) Failing after 1536h15m13s
CI/CD Pipeline / Staging API Integration Tests (pull_request) Failing after 1536h15m14s
CI/CD Pipeline / Staging E2E Tests (pull_request) Failing after 1536h15m14s
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (pull_request) Failing after 1536h15m16s
CI/CD Pipeline / Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Has been skipped
CI/CD Pipeline / Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Failing after 1536h46m52s
CI/CD Pipeline / Build Production Runtime Images (pull_request) Failing after 1536h46m53s

This commit is contained in:
2026-07-14 15:12:37 +08:00
parent 0f3881443f
commit 4b13efb488
+217 -135
View File
@@ -31,11 +31,21 @@ def main_audio_path(work_dir):
path = work_dir / "main.aac"
subprocess.run(
[
"ffmpeg", "-y", "-f", "lavfi",
"-i", "sine=frequency=440:duration=10:sample_rate=44100",
"-c:a", "aac", "-b:a", "128k", str(path),
"ffmpeg",
"-y",
"-f",
"lavfi",
"-i",
"sine=frequency=440:duration=10:sample_rate=44100",
"-c:a",
"aac",
"-b:a",
"128k",
str(path),
],
capture_output=True, check=True, timeout=30,
capture_output=True,
check=True,
timeout=30,
)
return path
@@ -48,11 +58,21 @@ def sfx_audio_path(work_dir):
path = work_dir / "sfx.aac"
subprocess.run(
[
"ffmpeg", "-y", "-f", "lavfi",
"-i", "sine=frequency=880:duration=3:sample_rate=44100",
"-c:a", "aac", "-b:a", "128k", str(path),
"ffmpeg",
"-y",
"-f",
"lavfi",
"-i",
"sine=frequency=880:duration=3:sample_rate=44100",
"-c:a",
"aac",
"-b:a",
"128k",
str(path),
],
capture_output=True, check=True, timeout=30,
capture_output=True,
check=True,
timeout=30,
)
return path
@@ -65,11 +85,21 @@ def voiceover_audio_path(work_dir):
path = work_dir / "voiceover.aac"
subprocess.run(
[
"ffmpeg", "-y", "-f", "lavfi",
"-i", "sine=frequency=220:duration=5:sample_rate=44100",
"-c:a", "aac", "-b:a", "128k", str(path),
"ffmpeg",
"-y",
"-f",
"lavfi",
"-i",
"sine=frequency=220:duration=5:sample_rate=44100",
"-c:a",
"aac",
"-b:a",
"128k",
str(path),
],
capture_output=True, check=True, timeout=30,
capture_output=True,
check=True,
timeout=30,
)
return path
@@ -82,15 +112,30 @@ def test_video_1(work_dir):
path = work_dir / "video1.mp4"
subprocess.run(
[
"ffmpeg", "-y", "-f", "lavfi",
"-i", "color=c=red:s=1080x1920:d=5:r=30",
"-f", "lavfi",
"-i", "sine=frequency=440:duration=5:sample_rate=44100",
"-c:v", "libx264", "-preset", "ultrafast",
"-c:a", "aac", "-b:a", "128k",
"-shortest", str(path),
"ffmpeg",
"-y",
"-f",
"lavfi",
"-i",
"color=c=red:s=1080x1920:d=5:r=30",
"-f",
"lavfi",
"-i",
"sine=frequency=440:duration=5:sample_rate=44100",
"-c:v",
"libx264",
"-preset",
"ultrafast",
"-c:a",
"aac",
"-b:a",
"128k",
"-shortest",
str(path),
],
capture_output=True, check=True, timeout=60,
capture_output=True,
check=True,
timeout=60,
)
return path
@@ -103,15 +148,30 @@ def test_video_2(work_dir):
path = work_dir / "video2.mp4"
subprocess.run(
[
"ffmpeg", "-y", "-f", "lavfi",
"-i", "color=c=blue:s=1080x1920:d=5:r=30",
"-f", "lavfi",
"-i", "sine=frequency=660:duration=5:sample_rate=44100",
"-c:v", "libx264", "-preset", "ultrafast",
"-c:a", "aac", "-b:a", "128k",
"-shortest", str(path),
"ffmpeg",
"-y",
"-f",
"lavfi",
"-i",
"color=c=blue:s=1080x1920:d=5:r=30",
"-f",
"lavfi",
"-i",
"sine=frequency=660:duration=5:sample_rate=44100",
"-c:v",
"libx264",
"-preset",
"ultrafast",
"-c:a",
"aac",
"-b:a",
"128k",
"-shortest",
str(path),
],
capture_output=True, check=True, timeout=60,
capture_output=True,
check=True,
timeout=60,
)
return path
@@ -139,42 +199,50 @@ class TestAudioTrack:
def test_volume_clamping(self):
from video_processing.multi_track_mixer import AudioTrack
track = AudioTrack.from_dict({
"track_id": "t1",
"audio_path": "/tmp/test.aac",
"volume": 5.0,
})
track = AudioTrack.from_dict(
{
"track_id": "t1",
"audio_path": "/tmp/test.aac",
"volume": 5.0,
}
)
assert track.volume == pytest.approx(2.0) # 上限钳制
track2 = AudioTrack.from_dict({
"track_id": "t2",
"audio_path": "/tmp/test.aac",
"volume": -1.0,
})
track2 = AudioTrack.from_dict(
{
"track_id": "t2",
"audio_path": "/tmp/test.aac",
"volume": -1.0,
}
)
assert track2.volume == pytest.approx(0.0) # 下限钳制
def test_track_type_default_volume(self):
from video_processing.multi_track_mixer import AudioTrack, DEFAULT_VOLUMES
from video_processing.multi_track_mixer import DEFAULT_VOLUMES, AudioTrack
for track_type, expected_vol in DEFAULT_VOLUMES.items():
track = AudioTrack.from_dict({
"track_id": "t1",
"track_type": track_type,
"audio_path": "/tmp/test.aac",
})
track = AudioTrack.from_dict(
{
"track_id": "t1",
"track_type": track_type,
"audio_path": "/tmp/test.aac",
}
)
assert track.volume == pytest.approx(expected_vol)
def test_invalid_config_safe(self):
from video_processing.multi_track_mixer import AudioTrack
# 无效值应该安全降级到默认值
track = AudioTrack.from_dict({
"track_id": "t1",
"audio_path": "/tmp/test.aac",
"volume": "invalid",
"fade_in": "abc",
"start_time": None,
})
track = AudioTrack.from_dict(
{
"track_id": "t1",
"audio_path": "/tmp/test.aac",
"volume": "invalid",
"fade_in": "abc",
"start_time": None,
}
)
assert track.volume > 0 # 有默认值
assert track.fade_in == 0.0
assert track.start_time == 0.0
@@ -182,11 +250,13 @@ class TestAudioTrack:
def test_disabled_track(self):
from video_processing.multi_track_mixer import AudioTrack
track = AudioTrack.from_dict({
"track_id": "t1",
"audio_path": "/tmp/test.aac",
"enabled": False,
})
track = AudioTrack.from_dict(
{
"track_id": "t1",
"audio_path": "/tmp/test.aac",
"enabled": False,
}
)
assert track.enabled is False
@@ -210,13 +280,15 @@ class TestMultiTrackMixConfig:
def test_valid_tracks(self):
from video_processing.multi_track_mixer import MultiTrackMixConfig
config = MultiTrackMixConfig.from_config_dict({
"tracks": [
{"track_id": "sfx1", "track_type": "sfx", "audio_path": "/tmp/sfx1.aac", "volume": 0.5},
{"track_id": "vo1", "track_type": "voiceover", "audio_path": "/tmp/vo1.aac"},
],
"master_volume": 0.8,
})
config = MultiTrackMixConfig.from_config_dict(
{
"tracks": [
{"track_id": "sfx1", "track_type": "sfx", "audio_path": "/tmp/sfx1.aac", "volume": 0.5},
{"track_id": "vo1", "track_type": "voiceover", "audio_path": "/tmp/vo1.aac"},
],
"master_volume": 0.8,
}
)
assert config.has_effect is True
assert len(config.tracks) == 2
assert config.tracks[0].volume == pytest.approx(0.5)
@@ -225,14 +297,16 @@ class TestMultiTrackMixConfig:
def test_skip_invalid_tracks(self):
from video_processing.multi_track_mixer import MultiTrackMixConfig
config = MultiTrackMixConfig.from_config_dict({
"tracks": [
{"track_id": "valid", "audio_path": "/tmp/valid.aac"},
{"track_id": "no_path"}, # 没有 audio_path,应该跳过
"not_a_dict", # 不是字典,应该跳过
{"track_id": "disabled", "audio_path": "/tmp/dis.aac", "enabled": False},
],
})
config = MultiTrackMixConfig.from_config_dict(
{
"tracks": [
{"track_id": "valid", "audio_path": "/tmp/valid.aac"},
{"track_id": "no_path"}, # 没有 audio_path,应该跳过
"not_a_dict", # 不是字典,应该跳过
{"track_id": "disabled", "audio_path": "/tmp/dis.aac", "enabled": False},
],
}
)
# 只有 valid 一个有效(disabled 的也跳过)
assert len([t for t in config.tracks if t.enabled]) == 1
@@ -394,7 +468,7 @@ class TestSubtitleStyle:
assert style3.alignment == 5
def test_9grid_positions(self):
from video_processing.subtitle_render_engine import SubtitleStyle, POSITION_ALIGNMENT
from video_processing.subtitle_render_engine import POSITION_ALIGNMENT, SubtitleStyle
for pos, align in POSITION_ALIGNMENT.items():
style = SubtitleStyle.from_dict({"position": pos})
@@ -410,11 +484,13 @@ class TestSubtitleStyle:
def test_background_style(self):
from video_processing.subtitle_render_engine import SubtitleStyle
style = SubtitleStyle.from_dict({
"background_enabled": True,
"background_color": "#000000",
"background_opacity": 0.7,
})
style = SubtitleStyle.from_dict(
{
"background_enabled": True,
"background_color": "#000000",
"background_opacity": 0.7,
}
)
assert style.background_enabled is True
assert style.background_opacity == pytest.approx(0.7)
@@ -429,11 +505,13 @@ class TestSubtitleStyle:
def test_safe_type_conversion(self):
from video_processing.subtitle_render_engine import SubtitleStyle
style = SubtitleStyle.from_dict({
"size": "invalid",
"margin_v": None,
"bold": "true", # 字符串真值
})
style = SubtitleStyle.from_dict(
{
"size": "invalid",
"margin_v": None,
"bold": "true", # 字符串真值
}
)
assert style.font_size == 24 # 降级到默认
assert style.margin_v == 60
# bool("true") = True,但这是 Python 行为,可以接受
@@ -511,10 +589,12 @@ class TestSubtitleRenderEngine:
engine = SubtitleRenderEngine(video_width=1080, video_height=1920, video_duration=10.0)
engine.add_title("视频标题")
engine.add_timeline_segments([
{"start": 0.0, "end": 3.0, "text": "ASR 结果1"},
{"start": 3.0, "end": 7.0, "text": "ASR 结果2"},
])
engine.add_timeline_segments(
[
{"start": 0.0, "end": 3.0, "text": "ASR 结果1"},
{"start": 3.0, "end": 7.0, "text": "ASR 结果2"},
]
)
output = work_dir / "mixed.ass"
engine.generate_ass(output)
@@ -595,9 +675,7 @@ class TestBuildSubtitlesFromPlan:
"style": {"size": 48, "bold": True, "position": "top"},
}
}
result = build_subtitles_from_plan(
output, config, video_width=1080, video_height=1920, video_duration=10.0
)
result = build_subtitles_from_plan(output, config, video_width=1080, video_height=1920, video_duration=10.0)
assert result is not None
assert result.exists()
content = result.read_text(encoding="utf-8")
@@ -614,9 +692,7 @@ class TestBuildSubtitlesFromPlan:
],
"manual_subtitle_style": {"size": 28, "color": "#FFFF00"},
}
result = build_subtitles_from_plan(
output, config, video_width=1080, video_height=1920, video_duration=10.0
)
result = build_subtitles_from_plan(output, config, video_width=1080, video_height=1920, video_duration=10.0)
assert result is not None
content = result.read_text(encoding="utf-8")
assert "手动字幕1" in content
@@ -633,9 +709,7 @@ class TestBuildSubtitlesFromPlan:
"text": "不显示的标题",
}
}
result = build_subtitles_from_plan(
output, config, video_width=1080, video_height=1920, video_duration=10.0
)
result = build_subtitles_from_plan(output, config, video_width=1080, video_height=1920, video_duration=10.0)
assert result is None
@@ -678,22 +752,26 @@ class TestConcatSegment:
def test_trimming_config(self):
from video_processing.concat_engine import ConcatSegment
seg = ConcatSegment.from_dict({
"video_path": "/tmp/test.mp4",
"start_time": 5.0,
"duration": 10.0,
})
seg = ConcatSegment.from_dict(
{
"video_path": "/tmp/test.mp4",
"start_time": 5.0,
"duration": 10.0,
}
)
assert seg.start_time == 5.0
assert seg.duration == 10.0
def test_invalid_values_safe(self):
from video_processing.concat_engine import ConcatSegment
seg = ConcatSegment.from_dict({
"video_path": "/tmp/test.mp4",
"start_time": "invalid",
"duration": -5.0,
})
seg = ConcatSegment.from_dict(
{
"video_path": "/tmp/test.mp4",
"start_time": "invalid",
"duration": -5.0,
}
)
assert seg.start_time == 0.0
assert seg.duration == 0.0
@@ -711,24 +789,28 @@ class TestConcatConfig:
def test_single_segment_no_effect(self):
from video_processing.concat_engine import ConcatConfig
config = ConcatConfig.from_config_dict({
"segments": [{"video_path": "/tmp/1.mp4"}],
})
config = ConcatConfig.from_config_dict(
{
"segments": [{"video_path": "/tmp/1.mp4"}],
}
)
assert config.has_effect is False # 只有一段不需要拼接
def test_multiple_segments(self):
from video_processing.concat_engine import ConcatConfig
config = ConcatConfig.from_config_dict({
"segments": [
{"video_path": "/tmp/1.mp4"},
{"video_path": "/tmp/2.mp4"},
{"video_path": "/tmp/3.mp4"},
],
"output_width": 1080,
"output_height": 1920,
"output_fps": 30.0,
})
config = ConcatConfig.from_config_dict(
{
"segments": [
{"video_path": "/tmp/1.mp4"},
{"video_path": "/tmp/2.mp4"},
{"video_path": "/tmp/3.mp4"},
],
"output_width": 1080,
"output_height": 1920,
"output_fps": 30.0,
}
)
assert config.has_effect is True
assert config.total_segments == 3
assert config.output_width == 1080
@@ -736,14 +818,16 @@ class TestConcatConfig:
def test_skip_invalid_segments(self):
from video_processing.concat_engine import ConcatConfig
config = ConcatConfig.from_config_dict({
"segments": [
{"video_path": "/tmp/1.mp4"},
{}, # 没有 path
{"video_path": ""}, # 空 path
{"video_path": "/tmp/2.mp4"},
],
})
config = ConcatConfig.from_config_dict(
{
"segments": [
{"video_path": "/tmp/1.mp4"},
{}, # 没有 path
{"video_path": ""}, # 空 path
{"video_path": "/tmp/2.mp4"},
],
}
)
assert config.total_segments == 2
assert config.has_effect is True
@@ -753,7 +837,7 @@ class TestConcatEngine:
def test_concat_demuxer_stream_copy(self, work_dir, test_video_1, test_video_2):
"""concat demuxer 模式(stream copy)."""
from video_processing.concat_engine import ConcatEngine, ConcatConfig, ConcatSegment
from video_processing.concat_engine import ConcatConfig, ConcatEngine, ConcatSegment
engine = ConcatEngine(work_dir=work_dir)
config = ConcatConfig(
@@ -777,7 +861,7 @@ class TestConcatEngine:
def test_concat_filter_reencode(self, work_dir, test_video_1, test_video_2):
"""concat filter 模式(强制重新编码)."""
from video_processing.concat_engine import ConcatEngine, ConcatConfig, ConcatSegment
from video_processing.concat_engine import ConcatConfig, ConcatEngine, ConcatSegment
engine = ConcatEngine(work_dir=work_dir)
config = ConcatConfig(
@@ -801,7 +885,7 @@ class TestConcatEngine:
def test_concat_with_trimming(self, work_dir, test_video_1, test_video_2):
"""带裁剪的拼接(自动用 filter 模式)."""
from video_processing.concat_engine import ConcatEngine, ConcatConfig, ConcatSegment
from video_processing.concat_engine import ConcatConfig, ConcatEngine, ConcatSegment
engine = ConcatEngine(work_dir=work_dir)
config = ConcatConfig(
@@ -824,7 +908,7 @@ class TestConcatEngine:
def test_single_segment_copy(self, work_dir, test_video_1):
"""单片段直接复制."""
from video_processing.concat_engine import ConcatEngine, ConcatConfig, ConcatSegment
from video_processing.concat_engine import ConcatConfig, ConcatEngine, ConcatSegment
engine = ConcatEngine(work_dir=work_dir)
config = ConcatConfig(
@@ -859,9 +943,7 @@ class TestConcatEngine:
{"video_path": str(test_video_2)},
],
}
result = concat_videos_from_config(
config_dict, output, work_dir=work_dir
)
result = concat_videos_from_config(config_dict, output, work_dir=work_dir)
assert result is not None
assert result.exists()
@@ -875,7 +957,7 @@ class TestConcatEngine:
def test_three_videos_concat(self, work_dir, test_video_1, test_video_2):
"""三段视频拼接."""
from video_processing.concat_engine import ConcatEngine, ConcatConfig, ConcatSegment
from video_processing.concat_engine import ConcatConfig, ConcatEngine, ConcatSegment
engine = ConcatEngine(work_dir=work_dir)
config = ConcatConfig(
@@ -899,7 +981,7 @@ class TestConcatEngine:
def test_output_resolution_override(self, work_dir, test_video_1, test_video_2):
"""指定输出分辨率."""
from video_processing.concat_engine import ConcatEngine, ConcatConfig, ConcatSegment
from video_processing.concat_engine import ConcatConfig, ConcatEngine, ConcatSegment
engine = ConcatEngine(work_dir=work_dir)
config = ConcatConfig(