| 1 | |
| 2 | import unittest |
| 3 | import os |
| 4 | import shutil |
| 5 | import sys |
| 6 | import tempfile |
| 7 | import types |
| 8 | from contextlib import redirect_stdout |
| 9 | from io import StringIO |
| 10 | from pathlib import Path |
| 11 | from unittest.mock import patch |
| 12 | from moviepy import ( |
| 13 | VideoFileClip, |
| 14 | ) |
| 15 | # add project root to python path |
| 16 | sys.path.insert(0, str(Path(__file__).parent.parent.parent)) |
| 17 | from app.config import config |
| 18 | from app.controllers.manager.base_manager import TaskQueueFullError |
| 19 | from app.controllers.manager.memory_manager import InMemoryTaskManager |
| 20 | from app.controllers.v1 import video as video_controller |
| 21 | from app.models import const |
| 22 | from app.models.schema import MaterialInfo |
| 23 | from app.services import state as sm |
| 24 | from app.services import video as vd |
| 25 | from app.utils import utils |
| 26 | |
| 27 | resources_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), "resources") |
| 28 | |
| 29 | |
| 30 | class _FakeRequest: |
| 31 | def __init__(self): |
| 32 | self.headers = {"x-task-id": "test-request"} |
| 33 | |
| 34 | |
| 35 | class TestSecurityControls(unittest.TestCase): |
| 36 | def setUp(self): |
| 37 | self.original_app_config = dict(config.app) |
| 38 | |
| 39 | def tearDown(self): |
| 40 | config.app.clear() |
| 41 | config.app.update(self.original_app_config) |
| 42 | |
| 43 | def test_task_query_returns_relative_task_url_without_mutating_state(self): |
| 44 | """ |
| 45 | endpoint 未显式配置时，任务查询接口不能使用 Host 派生绝对 URL， |
| 46 | 也不能把展示 URL 回写到任务状态里，否则不同 Host 查询会污染结果。 |
| 47 | """ |
| 48 | task_id = "security-task-url" |
| 49 | task_dir = utils.task_dir(task_id) |
| 50 | video_path = os.path.join(task_dir, "final-1.mp4") |
| 51 | Path(video_path).write_bytes(b"fake-video") |
| 52 | config.app["endpoint"] = "" |
| 53 | |
| 54 | try: |
| 55 | sm.state.update_task( |
| 56 | task_id, |
| 57 | state=const.TASK_STATE_COMPLETE, |
| 58 | videos=[video_path], |
| 59 | combined_videos=[video_path], |
| 60 | ) |
| 61 | |
| 62 | response = video_controller.get_task(_FakeRequest(), task_id=task_id) |
| 63 | |
| 64 | self.assertEqual(response["data"]["videos"], [f"/tasks/{task_id}/final-1.mp4"]) |
| 65 | self.assertEqual(sm.state.get_task(task_id)["videos"], [video_path]) |
| 66 | finally: |
| 67 | sm.state.delete_task(task_id) |
| 68 | shutil.rmtree(task_dir, ignore_errors=True) |
| 69 | |
| 70 | def test_in_memory_task_manager_rejects_when_queue_is_full(self): |
| 71 | """ |
| 72 | 并发数用尽后，等待队列必须有硬上限。这里用 max_concurrent_tasks=0 |
| 73 | 强制任务进入队列，验证超过 max_queued_tasks 时会拒绝继续入队。 |
| 74 | """ |
| 75 | manager = InMemoryTaskManager(max_concurrent_tasks=0, max_queued_tasks=1) |
| 76 | |
| 77 | manager.add_task(lambda: None) |
| 78 | |
| 79 | with self.assertRaises(TaskQueueFullError): |
| 80 | manager.add_task(lambda: None) |
| 81 | |
| 82 | class TestVideoService(unittest.TestCase): |
| 83 | def setUp(self): |
| 84 | self.original_app_config = dict(config.app) |
| 85 | self.test_img_path = os.path.join(resources_dir, "1.png") |
| 86 | vd._runtime_disabled_video_codecs.clear() |
| 87 | vd._ffmpeg_encoder_exists.cache_clear() |
| 88 | |
| 89 | def tearDown(self): |
| 90 | config.app.clear() |
| 91 | config.app.update(self.original_app_config) |
| 92 | vd._runtime_disabled_video_codecs.clear() |
| 93 | vd._ffmpeg_encoder_exists.cache_clear() |
| 94 | |
| 95 | def test_preprocess_video(self): |
| 96 | if not os.path.exists(self.test_img_path): |
| 97 | self.fail(f"test image not found: {self.test_img_path}") |
| 98 | |
| 99 | local_videos_dir = utils.storage_dir("local_videos", create=True) |
| 100 | safe_img_path = os.path.join(local_videos_dir, "test-preprocess-1.png") |
| 101 | shutil.copy2(self.test_img_path, safe_img_path) |
| 102 | |
| 103 | # test preprocess_video function |
| 104 | m = MaterialInfo() |
| 105 | m.url = os.path.basename(safe_img_path) |
| 106 | m.provider = "local" |
| 107 | print(m) |
| 108 | |
| 109 | try: |
| 110 | materials = vd.preprocess_video([m], clip_duration=4) |
| 111 | print(materials) |
| 112 | |
| 113 | # verify result |
| 114 | self.assertIsNotNone(materials) |
| 115 | self.assertEqual(len(materials), 1) |
| 116 | self.assertTrue(materials[0].url.endswith(".mp4")) |
| 117 | |
| 118 | # moviepy get video info |
| 119 | clip = VideoFileClip(materials[0].url) |
| 120 | try: |
| 121 | print(clip) |
| 122 | finally: |
| 123 | clip.close() |
| 124 | |
| 125 | # clean generated test video file |
| 126 | if os.path.exists(materials[0].url): |
| 127 | os.remove(materials[0].url) |
| 128 | finally: |
| 129 | if os.path.exists(safe_img_path): |
| 130 | os.remove(safe_img_path) |
| 131 | |
| 132 | def test_preprocess_video_rejects_material_outside_local_videos(self): |
| 133 | """ |
| 134 | local 素材路径来自 API 参数，不能允许任意绝对路径进入 MoviePy。 |
| 135 | 这里验证非 local_videos 白名单目录内的路径会被跳过，避免任意文件读取。 |
| 136 | """ |
| 137 | m = MaterialInfo(provider="local", url=self.test_img_path) |
| 138 | |
| 139 | materials = vd.preprocess_video([m], clip_duration=4) |
| 140 | |
| 141 | self.assertEqual(materials, []) |
| 142 | |
| 143 | def test_get_bgm_file_accepts_song_directory_filename(self): |
| 144 | """ |
| 145 | BGM 列表接口现在只暴露文件名；生成视频时应能把文件名安全解析回 |
| 146 | resource/songs 白名单目录，保持正常使用路径可用。 |
| 147 | """ |
| 148 | song_dir = utils.song_dir() |
| 149 | bgm_path = os.path.join(song_dir, "test-safe-bgm.mp3") |
| 150 | Path(bgm_path).write_bytes(b"fake-mp3") |
| 151 | |
| 152 | try: |
| 153 | self.assertEqual(vd.get_bgm_file(bgm_file="test-safe-bgm.mp3"), bgm_path) |
| 154 | finally: |
| 155 | if os.path.exists(bgm_path): |
| 156 | os.remove(bgm_path) |
| 157 | |
| 158 | def test_get_bgm_file_accepts_project_relative_song_path(self): |
| 159 | """ |
| 160 | 用户在 WebUI 中可能直接填写 ./resource/songs/xxx.mp3。该路径虽然是 |
| 161 | 项目根目录相对路径，但实际文件仍在 resource/songs 白名单目录内， |
| 162 | 应该被接受，避免自定义背景音乐被误判为不存在。 |
| 163 | """ |
| 164 | song_dir = utils.song_dir() |
| 165 | bgm_path = os.path.join(song_dir, "test-relative-bgm.mp3") |
| 166 | Path(bgm_path).write_bytes(b"fake-mp3") |
| 167 | |
| 168 | try: |
| 169 | self.assertEqual( |
| 170 | vd.get_bgm_file(bgm_file="./resource/songs/test-relative-bgm.mp3"), |
| 171 | bgm_path, |
| 172 | ) |
| 173 | finally: |
| 174 | if os.path.exists(bgm_path): |
| 175 | os.remove(bgm_path) |
| 176 | |
| 177 | def test_get_bgm_file_rejects_path_outside_song_directory(self): |
| 178 | """ |
| 179 | 用户传入的 bgm_file 不能直接作为本地路径打开，否则可能读取系统文件。 |
| 180 | 即使外部文件存在，也必须因为不在 songs 目录内被拒绝。 |
| 181 | """ |
| 182 | with tempfile.NamedTemporaryFile(suffix=".mp3") as temp_bgm: |
| 183 | self.assertEqual(vd.get_bgm_file(bgm_file=temp_bgm.name), "") |
| 184 | |
| 185 | def test_get_ffmpeg_binary_uses_configured_env_path(self): |
| 186 | """配置中显式指定 ffmpeg 时，应优先使用该路径。""" |
| 187 | with patch.dict(os.environ, {"IMAGEIO_FFMPEG_EXE": "/tmp/custom-ffmpeg"}, clear=True): |
| 188 | self.assertEqual(utils.get_ffmpeg_binary(), "/tmp/custom-ffmpeg") |
| 189 | |
| 190 | def test_get_ffmpeg_binary_falls_back_to_imageio_ffmpeg(self): |
| 191 | """ |
| 192 | Windows 便携包里系统 PATH 可能没有 ffmpeg，但 moviepy 依赖的 |
| 193 | imageio-ffmpeg 通常会提供可执行文件。这里验证该兜底路径可用。 |
| 194 | """ |
| 195 | fake_imageio_ffmpeg = types.SimpleNamespace( |
| 196 | get_ffmpeg_exe=lambda: "/tmp/bundled-ffmpeg" |
| 197 | ) |
| 198 | |
| 199 | with patch.dict(os.environ, {}, clear=True), patch.object( |
| 200 | utils.shutil, "which", return_value=None |
| 201 | ), patch.dict(sys.modules, {"imageio_ffmpeg": fake_imageio_ffmpeg}): |
| 202 | self.assertEqual(utils.get_ffmpeg_binary(), "/tmp/bundled-ffmpeg") |
| 203 | |
| 204 | def test_get_effective_video_codec_falls_back_when_encoder_missing(self): |
| 205 | """ |
| 206 | 用户选择的硬件编码器必须先经过 FFmpeg encoder 列表检测。检测不到 |
| 207 | 时直接回退 libx264，避免生成任务在写文件阶段才失败。 |
| 208 | """ |
| 209 | config.app["video_codec"] = "h264_nvenc" |
| 210 | |
| 211 | with patch.object(vd, "_ffmpeg_encoder_exists", return_value=False): |
| 212 | self.assertEqual(vd._get_effective_video_codec(), "libx264") |
| 213 | |
| 214 | def test_ffmpeg_encoder_exists_falls_back_when_probe_fails(self): |
| 215 | """ |
| 216 | Windows 上用户配置的 ffmpeg 可能因为路径损坏、权限或杀软拦截而无法 |
| 217 | 正常执行。encoder 探测失败时必须返回 False，让上层稳定回退 libx264。 |
| 218 | """ |
| 219 | with patch.object( |
| 220 | vd.subprocess, |
| 221 | "run", |
| 222 | side_effect=OSError("permission denied"), |
| 223 | ): |
| 224 | self.assertFalse(vd._ffmpeg_encoder_exists("C:/ffmpeg/bin/ffmpeg.exe", "h264_nvenc")) |
| 225 | |
| 226 | def test_write_videofile_falls_back_after_runtime_encoder_failure(self): |
| 227 | """ |
| 228 | FFmpeg 声明支持某个硬件编码器，不代表当前显卡或驱动一定可用。 |
| 229 | 首次实际编码失败后，应立即用 libx264 重试，并在本进程禁用该编码器。 |
| 230 | """ |
| 231 | |
| 232 | class _FakeClip: |
| 233 | def __init__(self): |
| 234 | self.codecs = [] |
| 235 | |
| 236 | def write_videofile(self, output_file, codec, **kwargs): |
| 237 | self.codecs.append(codec) |
| 238 | if codec == "h264_nvenc": |
| 239 | raise RuntimeError("nvenc device not available") |
| 240 | |
| 241 | fake_clip = _FakeClip() |
| 242 | |
| 243 | with patch.object(vd, "_ffmpeg_encoder_exists", return_value=True): |
| 244 | used_codec = vd._write_videofile_with_codec_fallback( |
| 245 | fake_clip, |
| 246 | "/tmp/fake.mp4", |
| 247 | codec="h264_nvenc", |
| 248 | logger=None, |
| 249 | fps=30, |
| 250 | ) |
| 251 | |
| 252 | self.assertEqual(used_codec, "libx264") |
| 253 | self.assertEqual(fake_clip.codecs, ["h264_nvenc", "libx264"]) |
| 254 | self.assertIn("h264_nvenc", vd._runtime_disabled_video_codecs) |
| 255 | |
| 256 | def test_write_videofile_does_not_disable_codec_when_fallback_also_fails(self): |
| 257 | """ |
| 258 | 如果 libx264 兜底也失败，失败原因更可能是输出路径、权限、文件占用等 |
| 259 | 通用问题，不能误判为硬件编码器不可用。 |
| 260 | """ |
| 261 | |
| 262 | class _FakeClip: |
| 263 | def write_videofile(self, output_file, codec, **kwargs): |
| 264 | raise RuntimeError(f"{codec} cannot write output") |
| 265 | |
| 266 | with patch.object(vd, "_ffmpeg_encoder_exists", return_value=True): |
| 267 | with self.assertRaises(RuntimeError): |
| 268 | vd._write_videofile_with_codec_fallback( |
| 269 | _FakeClip(), |
| 270 | "/tmp/fake.mp4", |
| 271 | codec="h264_nvenc", |
| 272 | logger=None, |
| 273 | fps=30, |
| 274 | ) |
| 275 | |
| 276 | self.assertNotIn("h264_nvenc", vd._runtime_disabled_video_codecs) |
| 277 | |
| 278 | def test_format_ffmpeg_concat_path_normalizes_windows_path(self): |
| 279 | """ |
| 280 | concat demuxer 的文件列表对 Windows 反斜杠较敏感，写入 list 前统一 |
| 281 | 转成正斜杠，并继续保留单引号转义。 |
| 282 | """ |
| 283 | with patch.object(vd.os.path, "abspath", return_value=r"C:\Users\Harry's Videos\clip.mp4"): |
| 284 | self.assertEqual( |
| 285 | vd._format_ffmpeg_concat_path(r"C:\Users\Harry's Videos\clip.mp4"), |
| 286 | "C:/Users/Harry'\\''s Videos/clip.mp4", |
| 287 | ) |
| 288 | |
| 289 | def test_concat_video_clips_falls_back_after_runtime_encoder_failure(self): |
| 290 | """ |
| 291 | 最终 ffmpeg concat 阶段也要具备同样的回退能力。这里用 mock 模拟 |
| 292 | h264_nvenc 编码失败，确认会自动再用 libx264 执行一次。 |
| 293 | """ |
| 294 | config.app["video_codec"] = "h264_nvenc" |
| 295 | |
| 296 | def fake_run(command, capture_output, text, check): |
| 297 | codec_index = command.index("-c:v") + 1 |
| 298 | codec = command[codec_index] |
| 299 | if codec == "h264_nvenc": |
| 300 | return types.SimpleNamespace( |
| 301 | returncode=1, |
| 302 | stdout="", |
| 303 | stderr="nvenc device not available", |
| 304 | ) |
| 305 | return types.SimpleNamespace(returncode=0, stdout="", stderr="") |
| 306 | |
| 307 | with tempfile.TemporaryDirectory() as temp_dir: |
| 308 | clip_file = os.path.join(temp_dir, "clip.mp4") |
| 309 | output_file = os.path.join(temp_dir, "combined.mp4") |
| 310 | Path(clip_file).write_bytes(b"fake") |
| 311 | |
| 312 | with patch.object(vd, "_ffmpeg_encoder_exists", return_value=True): |
| 313 | with patch.object(vd.subprocess, "run", side_effect=fake_run) as run: |
| 314 | vd.concat_video_clips_with_ffmpeg( |
| 315 | clip_files=[clip_file], |
| 316 | output_file=output_file, |
| 317 | threads=1, |
| 318 | output_dir=temp_dir, |
| 319 | ) |
| 320 | |
| 321 | used_codecs = [ |
| 322 | call.args[0][call.args[0].index("-c:v") + 1] |
| 323 | for call in run.call_args_list |
| 324 | ] |
| 325 | self.assertEqual(used_codecs, ["h264_nvenc", "libx264"]) |
| 326 | self.assertIn("h264_nvenc", vd._runtime_disabled_video_codecs) |
| 327 | |
| 328 | def test_concat_video_clips_does_not_disable_codec_when_fallback_also_fails(self): |
| 329 | """ |
| 330 | concat 阶段如果 libx264 也失败，说明可能是输入 list、路径或输出权限 |
| 331 | 问题，不能把硬件编码器加入运行时禁用列表。 |
| 332 | """ |
| 333 | config.app["video_codec"] = "h264_nvenc" |
| 334 | |
| 335 | def fake_run(command, capture_output, text, check): |
| 336 | codec_index = command.index("-c:v") + 1 |
| 337 | codec = command[codec_index] |
| 338 | return types.SimpleNamespace( |
| 339 | returncode=1, |
| 340 | stdout="", |
| 341 | stderr=f"{codec} cannot write output", |
| 342 | ) |
| 343 | |
| 344 | with tempfile.TemporaryDirectory() as temp_dir: |
| 345 | clip_file = os.path.join(temp_dir, "clip.mp4") |
| 346 | output_file = os.path.join(temp_dir, "combined.mp4") |
| 347 | Path(clip_file).write_bytes(b"fake") |
| 348 | |
| 349 | with patch.object(vd, "_ffmpeg_encoder_exists", return_value=True): |
| 350 | with patch.object(vd.subprocess, "run", side_effect=fake_run): |
| 351 | with self.assertRaises(RuntimeError): |
| 352 | vd.concat_video_clips_with_ffmpeg( |
| 353 | clip_files=[clip_file], |
| 354 | output_file=output_file, |
| 355 | threads=1, |
| 356 | output_dir=temp_dir, |
| 357 | ) |
| 358 | |
| 359 | self.assertNotIn("h264_nvenc", vd._runtime_disabled_video_codecs) |
| 360 | |
| 361 | def test_open_video_clip_quietly_suppresses_moviepy_stdout(self): |
| 362 | """ |
| 363 | MoviePy 2.1.x 的 FFMPEG_VideoReader 会直接向 stdout 打印 metadata |
| 364 | 和 ffmpeg 命令。项目服务层应屏蔽这类依赖库噪声，避免用户把 |
| 365 | `audio_found: False` 误判为最终视频没有音频。 |
| 366 | """ |
| 367 | video_path = os.path.join(resources_dir, "1.png.mp4") |
| 368 | if not os.path.exists(video_path): |
| 369 | self.fail(f"test video not found: {video_path}") |
| 370 | |
| 371 | stdout = StringIO() |
| 372 | with redirect_stdout(stdout): |
| 373 | clip = vd._open_video_clip_quietly(video_path) |
| 374 | |
| 375 | try: |
| 376 | self.assertEqual(stdout.getvalue(), "") |
| 377 | self.assertIsNone(clip.audio) |
| 378 | self.assertGreater(clip.duration, 0) |
| 379 | finally: |
| 380 | vd.close_clip(clip) |
| 381 | |
| 382 | def test_combine_videos_closes_audio_clip_when_duration_read_fails(self): |
| 383 | """ |
| 384 | `combine_videos()` 只需要读取旁白音频时长。即使读取 duration |
| 385 | 时发生异常，也必须关闭 AudioFileClip，避免文件句柄泄漏。 |
| 386 | """ |
| 387 | |
| 388 | class _FakeAudioReader: |
| 389 | def __init__(self): |
| 390 | self.closed = False |
| 391 | |
| 392 | def close(self): |
| 393 | self.closed = True |
| 394 | |
| 395 | class _BrokenAudioClip: |
| 396 | def __init__(self): |
| 397 | self.reader = _FakeAudioReader() |
| 398 | |
| 399 | @property |
| 400 | def duration(self): |
| 401 | raise RuntimeError("failed to read duration") |
| 402 | |
| 403 | fake_audio_clip = _BrokenAudioClip() |
| 404 | |
| 405 | with patch.object(vd, "AudioFileClip", return_value=fake_audio_clip): |
| 406 | with self.assertRaises(RuntimeError): |
| 407 | vd.combine_videos( |
| 408 | combined_video_path="/tmp/unused-combined.mp4", |
| 409 | video_paths=[], |
| 410 | audio_file="/tmp/unused-audio.mp3", |
| 411 | ) |
| 412 | |
| 413 | self.assertTrue(fake_audio_clip.reader.closed) |
| 414 | |
| 415 | def test_combine_videos_handles_none_transition_mode(self): |
| 416 | """ |
| 417 | Ensure `combine_videos` safely handles |
| 418 | `video_transition_mode=None`. |
| 419 | """ |
| 420 | class _FakeAudioClip: |
| 421 | @property |
| 422 | def duration(self): |
| 423 | return 10.0 |
| 424 | |
| 425 | def close(self): |
| 426 | pass |
| 427 | |
| 428 | with tempfile.TemporaryDirectory() as temp_dir: |
| 429 | combined_video_path = os.path.join(temp_dir, "combined.mp4") |
| 430 | audio_file = os.path.join(temp_dir, "audio.mp3") |
| 431 | |
| 432 | with patch.object(vd, "AudioFileClip", return_value=_FakeAudioClip()): |
| 433 | # Use empty video_paths to avoid heavy video processing while |
| 434 | # still exercising transition mode normalization logic. |
| 435 | result = vd.combine_videos( |
| 436 | combined_video_path=combined_video_path, |
| 437 | video_paths=[], |
| 438 | audio_file=audio_file, |
| 439 | video_transition_mode=None, |
| 440 | ) |
| 441 | self.assertEqual(result, combined_video_path) |
| 442 | |
| 443 | def test_combine_videos_keeps_small_duration_safety_margin(self): |
| 444 | """ |
| 445 | 音频和素材累计时长刚好相等时，仍应继续追加一个短片段作为安全余量。 |
| 446 | |
| 447 | FFmpeg 按帧率拼接后可能让最终视频比理论时长短几十毫秒。如果这里 |
| 448 | 在 10.0s == 10.0s 时立即停止，成片末尾就可能出现音频还在播放但 |
| 449 | 视频素材已经结束的边界问题。 |
| 450 | """ |
| 451 | |
| 452 | class _FakeAudioClip: |
| 453 | duration = 10.0 |
| 454 | |
| 455 | def close(self): |
| 456 | pass |
| 457 | |
| 458 | class _FakeVideoClip: |
| 459 | def __init__(self, duration): |
| 460 | self.duration = duration |
| 461 | self.size = (1080, 1920) |
| 462 | self.w = 1080 |
| 463 | self.h = 1920 |
| 464 | |
| 465 | def subclipped(self, start_time, end_time): |
| 466 | return _FakeVideoClip(end_time - start_time) |
| 467 | |
| 468 | video_durations = { |
| 469 | "clip-1.mp4": 3.0, |
| 470 | "clip-2.mp4": 4.0, |
| 471 | "clip-3.mp4": 3.0, |
| 472 | "clip-4.mp4": 2.0, |
| 473 | } |
| 474 | |
| 475 | def _open_fake_video_clip(video_path): |
| 476 | return _FakeVideoClip(video_durations[video_path]) |
| 477 | |
| 478 | with tempfile.TemporaryDirectory() as temp_dir: |
| 479 | combined_video_path = os.path.join(temp_dir, "combined.mp4") |
| 480 | |
| 481 | with patch.object(vd, "AudioFileClip", return_value=_FakeAudioClip()): |
| 482 | with patch.object( |
| 483 | vd, "_open_video_clip_quietly", side_effect=_open_fake_video_clip |
| 484 | ): |
| 485 | with patch.object( |
| 486 | vd, "_write_videofile_with_codec_fallback" |
| 487 | ) as write_mock: |
| 488 | with patch.object(vd, "concat_video_clips_with_ffmpeg"): |
| 489 | with patch.object(vd, "delete_files"): |
| 490 | result = vd.combine_videos( |
| 491 | combined_video_path=combined_video_path, |
| 492 | video_paths=list(video_durations.keys()), |
| 493 | audio_file=os.path.join(temp_dir, "audio.mp3"), |
| 494 | video_aspect=vd.VideoAspect.portrait, |
| 495 | video_concat_mode=vd.VideoConcatMode.sequential, |
| 496 | video_transition_mode=None, |
| 497 | max_clip_duration=10, |
| 498 | ) |
| 499 | |
| 500 | self.assertEqual(result, combined_video_path) |
| 501 | self.assertEqual(write_mock.call_count, 4) |
| 502 | |
| 503 | def test_prioritize_unique_source_clips_uses_each_source_before_reuse(self): |
| 504 | """ |
| 505 | 随机模式下，一个长素材会被拆成多个片段。调度层应先让每个源素材 |
| 506 | 至少出现一次，再使用同一源素材的其他切片，降低用户感知到的重复。 |
| 507 | """ |
| 508 | clips = [ |
| 509 | vd.SubClippedVideoClip("a.mp4", 0, 4, source_file_path="a.mp4"), |
| 510 | vd.SubClippedVideoClip("a.mp4", 4, 8, source_file_path="a.mp4"), |
| 511 | vd.SubClippedVideoClip("b.mp4", 0, 4, source_file_path="b.mp4"), |
| 512 | vd.SubClippedVideoClip("b.mp4", 4, 8, source_file_path="b.mp4"), |
| 513 | vd.SubClippedVideoClip("c.mp4", 0, 4, source_file_path="c.mp4"), |
| 514 | ] |
| 515 | |
| 516 | ordered_clips = vd._prioritize_unique_source_clips( |
| 517 | subclipped_items=clips, |
| 518 | concat_mode=vd.VideoConcatMode.random, |
| 519 | ) |
| 520 | |
| 521 | self.assertCountEqual(ordered_clips, clips) |
| 522 | first_round_sources = [clip.source_file_path for clip in ordered_clips[:3]] |
| 523 | self.assertCountEqual(first_round_sources, ["a.mp4", "b.mp4", "c.mp4"]) |
| 524 | |
| 525 | def test_prioritize_unique_source_clips_keeps_sequential_order(self): |
| 526 | """ |
| 527 | 顺序模式本身只取每个素材的首段，不应被随机调度逻辑改变顺序。 |
| 528 | """ |
| 529 | clips = [ |
| 530 | vd.SubClippedVideoClip("a.mp4", 0, 4, source_file_path="a.mp4"), |
| 531 | vd.SubClippedVideoClip("b.mp4", 0, 4, source_file_path="b.mp4"), |
| 532 | vd.SubClippedVideoClip("c.mp4", 0, 4, source_file_path="c.mp4"), |
| 533 | ] |
| 534 | |
| 535 | ordered_clips = vd._prioritize_unique_source_clips( |
| 536 | subclipped_items=clips, |
| 537 | concat_mode=vd.VideoConcatMode.sequential, |
| 538 | ) |
| 539 | |
| 540 | self.assertEqual(ordered_clips, clips) |
| 541 | |
| 542 | def test_prioritize_unique_source_clips_prefers_long_primary_clip(self): |
| 543 | """ |
| 544 | 同一个源素材的最后一个切片可能短于目标片段时长。首轮去重时应优先 |
| 545 | 选择较长片段，否则会因为累计时长不足而提前复用素材。 |
| 546 | """ |
| 547 | short_tail = vd.SubClippedVideoClip( |
| 548 | "a.mp4", 6, 6.5, source_file_path="a.mp4" |
| 549 | ) |
| 550 | full_clip = vd.SubClippedVideoClip( |
| 551 | "a.mp4", 0, 3, source_file_path="a.mp4" |
| 552 | ) |
| 553 | other_source = vd.SubClippedVideoClip( |
| 554 | "b.mp4", 0, 3, source_file_path="b.mp4" |
| 555 | ) |
| 556 | |
| 557 | ordered_clips = vd._prioritize_unique_source_clips( |
| 558 | subclipped_items=[short_tail, full_clip, other_source], |
| 559 | concat_mode=vd.VideoConcatMode.random, |
| 560 | ) |
| 561 | |
| 562 | first_a_clip = next( |
| 563 | clip for clip in ordered_clips if clip.source_file_path == "a.mp4" |
| 564 | ) |
| 565 | self.assertEqual(first_a_clip, full_clip) |
| 566 | |
| 567 | def test_wrap_text(self): |
| 568 | """test text wrapping function""" |
| 569 | try: |
| 570 | font_path = os.path.join(utils.font_dir(), "STHeitiMedium.ttc") |
| 571 | if not os.path.exists(font_path): |
| 572 | self.fail(f"font file not found: {font_path}") |
| 573 | |
| 574 | # test english text wrapping |
| 575 | test_text_en = "This is a test text for wrapping long sentences in english language" |
| 576 | |
| 577 | wrapped_text_en, text_height_en = vd.wrap_text( |
| 578 | text=test_text_en, |
| 579 | max_width=300, |
| 580 | font=font_path, |
| 581 | fontsize=30 |
| 582 | ) |
| 583 | print(wrapped_text_en, text_height_en) |
| 584 | # verify text is wrapped |
| 585 | self.assertIn("\n", wrapped_text_en) |
| 586 | |
| 587 | # test chinese text wrapping |
| 588 | test_text_zh = "这是一段用来测试中文长句换行的文本内容，应该会根据宽度限制进行换行处理" |
| 589 | wrapped_text_zh, text_height_zh = vd.wrap_text( |
| 590 | text=test_text_zh, |
| 591 | max_width=300, |
| 592 | font=font_path, |
| 593 | fontsize=30 |
| 594 | ) |
| 595 | print(wrapped_text_zh, text_height_zh) |
| 596 | # verify chinese text is wrapped |
| 597 | self.assertIn("\n", wrapped_text_zh) |
| 598 | except Exception as e: |
| 599 | self.fail(f"test wrap_text failed: {str(e)}") |
| 600 | |
| 601 | def test_rounded_subtitle_background_clip_has_transparent_corners(self): |
| 602 | """ |
| 603 | 圆角字幕背景只在用户显式开启时使用。这里直接验证生成的 RGBA |
| 604 | 背景具备透明圆角和半透明中心，避免后续改动把圆角效果退化成实心矩形。 |
| 605 | """ |
| 606 | clip = vd._rounded_subtitle_background_clip( |
| 607 | width=120, |
| 608 | height=48, |
| 609 | color="#123456", |
| 610 | alpha=140, |
| 611 | radius=16, |
| 612 | ) |
| 613 | try: |
| 614 | frame = clip.get_frame(0) |
| 615 | mask = clip.mask.get_frame(0) |
| 616 | |
| 617 | self.assertEqual(frame.shape[0:2], (48, 120)) |
| 618 | self.assertEqual(tuple(frame[24, 60]), (18, 52, 86)) |
| 619 | self.assertEqual(mask[0, 0], 0) |
| 620 | self.assertGreater(mask[24, 60], 0.5) |
| 621 | self.assertLess(mask[24, 60], 0.6) |
| 622 | finally: |
| 623 | clip.close() |
| 624 | |
| 625 | def test_get_temp_audio_dir_returns_system_temp_on_windows(self): |
| 626 | with patch("sys.platform", "win32"): |
| 627 | result = vd._get_temp_audio_dir("/some/output/dir") |
| 628 | self.assertEqual(result, tempfile.gettempdir()) |
| 629 | |
| 630 | def test_get_temp_audio_dir_returns_output_dir_on_non_windows(self): |
| 631 | for platform in ("linux", "darwin"): |
| 632 | with self.subTest(platform=platform): |
| 633 | with patch("sys.platform", platform): |
| 634 | result = vd._get_temp_audio_dir("/some/output/dir") |
| 635 | self.assertEqual(result, "/some/output/dir") |
| 636 | |
| 637 | |
| 638 | if __name__ == "__main__": |
| 639 | unittest.main() |
| 640 |