import unittest import os import shutil import sys import tempfile import types from contextlib import redirect_stdout from io import StringIO from pathlib import Path from unittest.mock import patch from moviepy import ( VideoFileClip, ) # add project root to python path sys.path.insert(0, str(Path(__file__).parent.parent.parent)) from app.config import config from app.controllers.manager.base_manager import TaskQueueFullError from app.controllers.manager.memory_manager import InMemoryTaskManager from app.controllers.v1 import video as video_controller from app.models import const from app.models.schema import MaterialInfo from app.services import state as sm from app.services import video as vd from app.utils import utils resources_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), "resources") class _FakeRequest: def __init__(self): self.headers = {"x-task-id": "test-request"} class TestSecurityControls(unittest.TestCase): def setUp(self): self.original_app_config = dict(config.app) def tearDown(self): config.app.clear() config.app.update(self.original_app_config) def test_task_query_returns_relative_task_url_without_mutating_state(self): """ endpoint 未显式配置时，任务查询接口不能使用 Host 派生绝对 URL， 也不能把展示 URL 回写到任务状态里，否则不同 Host 查询会污染结果。 """ task_id = "security-task-url" task_dir = utils.task_dir(task_id) video_path = os.path.join(task_dir, "final-1.mp4") Path(video_path).write_bytes(b"fake-video") config.app["endpoint"] = "" try: sm.state.update_task( task_id, state=const.TASK_STATE_COMPLETE, videos=[video_path], combined_videos=[video_path], ) response = video_controller.get_task(_FakeRequest(), task_id=task_id) self.assertEqual(response["data"]["videos"], [f"/tasks/{task_id}/final-1.mp4"]) self.assertEqual(sm.state.get_task(task_id)["videos"], [video_path]) finally: sm.state.delete_task(task_id) shutil.rmtree(task_dir, ignore_errors=True) def test_in_memory_task_manager_rejects_when_queue_is_full(self): """ 并发数用尽后，等待队列必须有硬上限。这里用 max_concurrent_tasks=0 强制任务进入队列，验证超过 max_queued_tasks 时会拒绝继续入队。 """ manager = InMemoryTaskManager(max_concurrent_tasks=0, max_queued_tasks=1) manager.add_task(lambda: None) with self.assertRaises(TaskQueueFullError): manager.add_task(lambda: None) class TestVideoService(unittest.TestCase): def setUp(self): self.original_app_config = dict(config.app) self.test_img_path = os.path.join(resources_dir, "1.png") vd._runtime_disabled_video_codecs.clear() vd._ffmpeg_encoder_exists.cache_clear() def tearDown(self): config.app.clear() config.app.update(self.original_app_config) vd._runtime_disabled_video_codecs.clear() vd._ffmpeg_encoder_exists.cache_clear() def test_preprocess_video(self): if not os.path.exists(self.test_img_path): self.fail(f"test image not found: {self.test_img_path}") local_videos_dir = utils.storage_dir("local_videos", create=True) safe_img_path = os.path.join(local_videos_dir, "test-preprocess-1.png") shutil.copy2(self.test_img_path, safe_img_path) # test preprocess_video function m = MaterialInfo() m.url = os.path.basename(safe_img_path) m.provider = "local" print(m) try: materials = vd.preprocess_video([m], clip_duration=4) print(materials) # verify result self.assertIsNotNone(materials) self.assertEqual(len(materials), 1) self.assertTrue(materials[0].url.endswith(".mp4")) # moviepy get video info clip = VideoFileClip(materials[0].url) try: print(clip) finally: clip.close() # clean generated test video file if os.path.exists(materials[0].url): os.remove(materials[0].url) finally: if os.path.exists(safe_img_path): os.remove(safe_img_path) def test_preprocess_video_rejects_material_outside_local_videos(self): """ local 素材路径来自 API 参数，不能允许任意绝对路径进入 MoviePy。 这里验证非 local_videos 白名单目录内的路径会被跳过，避免任意文件读取。 """ m = MaterialInfo(provider="local", url=self.test_img_path) materials = vd.preprocess_video([m], clip_duration=4) self.assertEqual(materials, []) def test_get_bgm_file_accepts_song_directory_filename(self): """ BGM 列表接口现在只暴露文件名；生成视频时应能把文件名安全解析回 resource/songs 白名单目录，保持正常使用路径可用。 """ song_dir = utils.song_dir() bgm_path = os.path.join(song_dir, "test-safe-bgm.mp3") Path(bgm_path).write_bytes(b"fake-mp3") try: self.assertEqual(vd.get_bgm_file(bgm_file="test-safe-bgm.mp3"), bgm_path) finally: if os.path.exists(bgm_path): os.remove(bgm_path) def test_get_bgm_file_accepts_project_relative_song_path(self): """ 用户在 WebUI 中可能直接填写 ./resource/songs/xxx.mp3。该路径虽然是 项目根目录相对路径，但实际文件仍在 resource/songs 白名单目录内， 应该被接受，避免自定义背景音乐被误判为不存在。 """ song_dir = utils.song_dir() bgm_path = os.path.join(song_dir, "test-relative-bgm.mp3") Path(bgm_path).write_bytes(b"fake-mp3") try: self.assertEqual( vd.get_bgm_file(bgm_file="./resource/songs/test-relative-bgm.mp3"), bgm_path, ) finally: if os.path.exists(bgm_path): os.remove(bgm_path) def test_get_bgm_file_rejects_path_outside_song_directory(self): """ 用户传入的 bgm_file 不能直接作为本地路径打开，否则可能读取系统文件。 即使外部文件存在，也必须因为不在 songs 目录内被拒绝。 """ with tempfile.NamedTemporaryFile(suffix=".mp3") as temp_bgm: self.assertEqual(vd.get_bgm_file(bgm_file=temp_bgm.name), "") def test_get_ffmpeg_binary_uses_configured_env_path(self): """配置中显式指定 ffmpeg 时，应优先使用该路径。""" with patch.dict(os.environ, {"IMAGEIO_FFMPEG_EXE": "/tmp/custom-ffmpeg"}, clear=True): self.assertEqual(utils.get_ffmpeg_binary(), "/tmp/custom-ffmpeg") def test_get_ffmpeg_binary_falls_back_to_imageio_ffmpeg(self): """ Windows 便携包里系统 PATH 可能没有 ffmpeg，但 moviepy 依赖的 imageio-ffmpeg 通常会提供可执行文件。这里验证该兜底路径可用。 """ fake_imageio_ffmpeg = types.SimpleNamespace( get_ffmpeg_exe=lambda: "/tmp/bundled-ffmpeg" ) with patch.dict(os.environ, {}, clear=True), patch.object( utils.shutil, "which", return_value=None ), patch.dict(sys.modules, {"imageio_ffmpeg": fake_imageio_ffmpeg}): self.assertEqual(utils.get_ffmpeg_binary(), "/tmp/bundled-ffmpeg") def test_get_effective_video_codec_falls_back_when_encoder_missing(self): """ 用户选择的硬件编码器必须先经过 FFmpeg encoder 列表检测。检测不到 时直接回退 libx264，避免生成任务在写文件阶段才失败。 """ config.app["video_codec"] = "h264_nvenc" with patch.object(vd, "_ffmpeg_encoder_exists", return_value=False): self.assertEqual(vd._get_effective_video_codec(), "libx264") def test_get_configured_video_codec_uses_stable_default_when_unset(self): """ WebUI 的“默认”模式不会持久化 video_codec。后端必须在配置缺失时继续 明确返回 libx264，不能把空值直接交给 MoviePy 或 FFmpeg 自行决定。 """ config.app.pop("video_codec", None) self.assertEqual(vd._get_configured_video_codec(), "libx264") def test_get_configured_video_codec_preserves_explicit_libx264(self): """ 用户明确选择 libx264 时需要保持固定选择。它与“跟随项目默认策略”当前 结果相同，但配置语义不同，未来调整默认值时不能影响显式选择。 """ config.app["video_codec"] = "libx264" self.assertEqual(vd._get_configured_video_codec(), "libx264") def test_ffmpeg_encoder_exists_falls_back_when_probe_fails(self): """ Windows 上用户配置的 ffmpeg 可能因为路径损坏、权限或杀软拦截而无法 正常执行。encoder 探测失败时必须返回 False，让上层稳定回退 libx264。 """ with patch.object( vd.subprocess, "run", side_effect=OSError("permission denied"), ): self.assertFalse(vd._ffmpeg_encoder_exists("C:/ffmpeg/bin/ffmpeg.exe", "h264_nvenc")) def test_write_videofile_falls_back_after_runtime_encoder_failure(self): """ FFmpeg 声明支持某个硬件编码器，不代表当前显卡或驱动一定可用。 首次实际编码失败后，应立即用 libx264 重试，并在本进程禁用该编码器。 """ class _FakeClip: def __init__(self): self.codecs = [] def write_videofile(self, output_file, codec, **kwargs): self.codecs.append(codec) if codec == "h264_nvenc": raise RuntimeError("nvenc device not available") fake_clip = _FakeClip() with patch.object(vd, "_ffmpeg_encoder_exists", return_value=True): used_codec = vd._write_videofile_with_codec_fallback( fake_clip, "/tmp/fake.mp4", codec="h264_nvenc", logger=None, fps=30, ) self.assertEqual(used_codec, "libx264") self.assertEqual(fake_clip.codecs, ["h264_nvenc", "libx264"]) self.assertIn("h264_nvenc", vd._runtime_disabled_video_codecs) def test_write_videofile_does_not_disable_codec_when_fallback_also_fails(self): """ 如果 libx264 兜底也失败，失败原因更可能是输出路径、权限、文件占用等 通用问题，不能误判为硬件编码器不可用。 """ class _FakeClip: def write_videofile(self, output_file, codec, **kwargs): raise RuntimeError(f"{codec} cannot write output") with patch.object(vd, "_ffmpeg_encoder_exists", return_value=True): with self.assertRaises(RuntimeError): vd._write_videofile_with_codec_fallback( _FakeClip(), "/tmp/fake.mp4", codec="h264_nvenc", logger=None, fps=30, ) self.assertNotIn("h264_nvenc", vd._runtime_disabled_video_codecs) def test_format_ffmpeg_concat_path_normalizes_windows_path(self): """ concat demuxer 的文件列表对 Windows 反斜杠较敏感，写入 list 前统一 转成正斜杠，并继续保留单引号转义。 """ with patch.object( vd.os.path, "abspath", return_value=r"C:\Users\Test User's Videos\clip.mp4", ): self.assertEqual( vd._format_ffmpeg_concat_path( r"C:\Users\Test User's Videos\clip.mp4" ), "C:/Users/Test User'\\''s Videos/clip.mp4", ) def test_concat_video_clips_falls_back_after_runtime_encoder_failure(self): """ 最终 ffmpeg concat 阶段也要具备同样的回退能力。这里用 mock 模拟 h264_nvenc 编码失败，确认会自动再用 libx264 执行一次。 """ config.app["video_codec"] = "h264_nvenc" def fake_run(command, capture_output, text, check): codec_index = command.index("-c:v") + 1 codec = command[codec_index] if codec == "h264_nvenc": return types.SimpleNamespace( returncode=1, stdout="", stderr="nvenc device not available", ) return types.SimpleNamespace(returncode=0, stdout="", stderr="") with tempfile.TemporaryDirectory() as temp_dir: clip_file = os.path.join(temp_dir, "clip.mp4") output_file = os.path.join(temp_dir, "combined.mp4") Path(clip_file).write_bytes(b"fake") with patch.object(vd, "_ffmpeg_encoder_exists", return_value=True): with patch.object(vd.subprocess, "run", side_effect=fake_run) as run: vd.concat_video_clips_with_ffmpeg( clip_files=[clip_file], output_file=output_file, threads=1, output_dir=temp_dir, ) used_codecs = [ call.args[0][call.args[0].index("-c:v") + 1] for call in run.call_args_list ] self.assertEqual(used_codecs, ["h264_nvenc", "libx264"]) self.assertIn("h264_nvenc", vd._runtime_disabled_video_codecs) def test_concat_video_clips_does_not_disable_codec_when_fallback_also_fails(self): """ concat 阶段如果 libx264 也失败，说明可能是输入 list、路径或输出权限 问题，不能把硬件编码器加入运行时禁用列表。 """ config.app["video_codec"] = "h264_nvenc" def fake_run(command, capture_output, text, check): codec_index = command.index("-c:v") + 1 codec = command[codec_index] return types.SimpleNamespace( returncode=1, stdout="", stderr=f"{codec} cannot write output", ) with tempfile.TemporaryDirectory() as temp_dir: clip_file = os.path.join(temp_dir, "clip.mp4") output_file = os.path.join(temp_dir, "combined.mp4") Path(clip_file).write_bytes(b"fake") with patch.object(vd, "_ffmpeg_encoder_exists", return_value=True): with patch.object(vd.subprocess, "run", side_effect=fake_run): with self.assertRaises(RuntimeError): vd.concat_video_clips_with_ffmpeg( clip_files=[clip_file], output_file=output_file, threads=1, output_dir=temp_dir, ) self.assertNotIn("h264_nvenc", vd._runtime_disabled_video_codecs) def test_open_video_clip_quietly_suppresses_moviepy_stdout(self): """ MoviePy 2.1.x 的 FFMPEG_VideoReader 会直接向 stdout 打印 metadata 和 ffmpeg 命令。项目服务层应屏蔽这类依赖库噪声，避免用户把 `audio_found: False` 误判为最终视频没有音频。 """ video_path = os.path.join(resources_dir, "1.png.mp4") if not os.path.exists(video_path): self.fail(f"test video not found: {video_path}") stdout = StringIO() with redirect_stdout(stdout): clip = vd._open_video_clip_quietly(video_path) try: self.assertEqual(stdout.getvalue(), "") self.assertIsNone(clip.audio) self.assertGreater(clip.duration, 0) finally: vd.close_clip(clip) def test_combine_videos_closes_audio_clip_when_duration_read_fails(self): """ `combine_videos()` 只需要读取旁白音频时长。即使读取 duration 时发生异常，也必须关闭 AudioFileClip，避免文件句柄泄漏。 """ class _FakeAudioReader: def __init__(self): self.closed = False def close(self): self.closed = True class _BrokenAudioClip: def __init__(self): self.reader = _FakeAudioReader() @property def duration(self): raise RuntimeError("failed to read duration") fake_audio_clip = _BrokenAudioClip() with patch.object(vd, "AudioFileClip", return_value=fake_audio_clip): with self.assertRaises(RuntimeError): vd.combine_videos( combined_video_path="/tmp/unused-combined.mp4", video_paths=[], audio_file="/tmp/unused-audio.mp3", ) self.assertTrue(fake_audio_clip.reader.closed) def test_combine_videos_handles_none_transition_mode(self): """ Ensure `combine_videos` safely handles `video_transition_mode=None`. """ class _FakeAudioClip: @property def duration(self): return 10.0 def close(self): pass with tempfile.TemporaryDirectory() as temp_dir: combined_video_path = os.path.join(temp_dir, "combined.mp4") audio_file = os.path.join(temp_dir, "audio.mp3") with patch.object(vd, "AudioFileClip", return_value=_FakeAudioClip()): # Use empty video_paths to avoid heavy video processing while # still exercising transition mode normalization logic. result = vd.combine_videos( combined_video_path=combined_video_path, video_paths=[], audio_file=audio_file, video_transition_mode=None, ) self.assertEqual(result, combined_video_path) def test_combine_videos_keeps_small_duration_safety_margin(self): """ 音频和素材累计时长刚好相等时，仍应继续追加一个短片段作为安全余量。 FFmpeg 按帧率拼接后可能让最终视频比理论时长短几十毫秒。如果这里 在 10.0s == 10.0s 时立即停止，成片末尾就可能出现音频还在播放但 视频素材已经结束的边界问题。 """ class _FakeAudioClip: duration = 10.0 def close(self): pass class _FakeVideoClip: def __init__(self, duration): self.duration = duration self.size = (1080, 1920) self.w = 1080 self.h = 1920 def subclipped(self, start_time, end_time): return _FakeVideoClip(end_time - start_time) video_durations = { "clip-1.mp4": 3.0, "clip-2.mp4": 4.0, "clip-3.mp4": 3.0, "clip-4.mp4": 2.0, } def _open_fake_video_clip(video_path): return _FakeVideoClip(video_durations[video_path]) with tempfile.TemporaryDirectory() as temp_dir: combined_video_path = os.path.join(temp_dir, "combined.mp4") with patch.object(vd, "AudioFileClip", return_value=_FakeAudioClip()): with patch.object( vd, "_open_video_clip_quietly", side_effect=_open_fake_video_clip ): with patch.object( vd, "_write_videofile_with_codec_fallback" ) as write_mock: with patch.object(vd, "concat_video_clips_with_ffmpeg") as concat_mock: with patch.object(vd, "delete_files"): result = vd.combine_videos( combined_video_path=combined_video_path, video_paths=list(video_durations.keys()), audio_file=os.path.join(temp_dir, "audio.mp3"), video_aspect=vd.VideoAspect.portrait, video_concat_mode=vd.VideoConcatMode.sequential, video_transition_mode=None, max_clip_duration=10, ) self.assertEqual(result, combined_video_path) self.assertEqual(write_mock.call_count, 4) self.assertEqual(concat_mock.call_args.kwargs["max_duration"], 10.0) def test_concat_video_clips_limits_output_to_audio_duration(self): """最终拼接时应裁到音频时长，避免安全余量带来明显静音尾巴。""" def fake_run(command, capture_output, text, check): return types.SimpleNamespace(returncode=0, stdout="", stderr="") with tempfile.TemporaryDirectory() as temp_dir: clip_file = os.path.join(temp_dir, "clip.mp4") output_file = os.path.join(temp_dir, "combined.mp4") Path(clip_file).write_bytes(b"fake") with patch.object(vd.subprocess, "run", side_effect=fake_run) as run: vd.concat_video_clips_with_ffmpeg( clip_files=[clip_file], output_file=output_file, threads=1, output_dir=temp_dir, max_duration=10.0, ) command = run.call_args.args[0] self.assertEqual(command[command.index("-t") + 1], "10.000") self.assertLess(command.index("-t"), command.index(output_file)) def test_prioritize_unique_source_clips_uses_each_source_before_reuse(self): """ 随机模式下，一个长素材会被拆成多个片段。调度层应先让每个源素材 至少出现一次，再使用同一源素材的其他切片，降低用户感知到的重复。 """ clips = [ vd.SubClippedVideoClip("a.mp4", 0, 4, source_file_path="a.mp4"), vd.SubClippedVideoClip("a.mp4", 4, 8, source_file_path="a.mp4"), vd.SubClippedVideoClip("b.mp4", 0, 4, source_file_path="b.mp4"), vd.SubClippedVideoClip("b.mp4", 4, 8, source_file_path="b.mp4"), vd.SubClippedVideoClip("c.mp4", 0, 4, source_file_path="c.mp4"), ] ordered_clips = vd._prioritize_unique_source_clips( subclipped_items=clips, concat_mode=vd.VideoConcatMode.random, ) self.assertCountEqual(ordered_clips, clips) first_round_sources = [clip.source_file_path for clip in ordered_clips[:3]] self.assertCountEqual(first_round_sources, ["a.mp4", "b.mp4", "c.mp4"]) def test_prioritize_unique_source_clips_keeps_sequential_order(self): """ 顺序模式本身只取每个素材的首段，不应被随机调度逻辑改变顺序。 """ clips = [ vd.SubClippedVideoClip("a.mp4", 0, 4, source_file_path="a.mp4"), vd.SubClippedVideoClip("b.mp4", 0, 4, source_file_path="b.mp4"), vd.SubClippedVideoClip("c.mp4", 0, 4, source_file_path="c.mp4"), ] ordered_clips = vd._prioritize_unique_source_clips( subclipped_items=clips, concat_mode=vd.VideoConcatMode.sequential, ) self.assertEqual(ordered_clips, clips) def test_prioritize_unique_source_clips_prefers_long_primary_clip(self): """ 同一个源素材的最后一个切片可能短于目标片段时长。首轮去重时应优先 选择较长片段，否则会因为累计时长不足而提前复用素材。 """ short_tail = vd.SubClippedVideoClip( "a.mp4", 6, 6.5, source_file_path="a.mp4" ) full_clip = vd.SubClippedVideoClip( "a.mp4", 0, 3, source_file_path="a.mp4" ) other_source = vd.SubClippedVideoClip( "b.mp4", 0, 3, source_file_path="b.mp4" ) ordered_clips = vd._prioritize_unique_source_clips( subclipped_items=[short_tail, full_clip, other_source], concat_mode=vd.VideoConcatMode.random, ) first_a_clip = next( clip for clip in ordered_clips if clip.source_file_path == "a.mp4" ) self.assertEqual(first_a_clip, full_clip) def test_wrap_text(self): """test text wrapping function""" try: font_path = os.path.join(utils.font_dir(), "STHeitiMedium.ttc") if not os.path.exists(font_path): self.fail(f"font file not found: {font_path}") # test english text wrapping test_text_en = "This is a test text for wrapping long sentences in english language" wrapped_text_en, text_height_en = vd.wrap_text( text=test_text_en, max_width=300, font=font_path, fontsize=30 ) print(wrapped_text_en, text_height_en) # verify text is wrapped self.assertIn("\n", wrapped_text_en) # test chinese text wrapping test_text_zh = "这是一段用来测试中文长句换行的文本内容，应该会根据宽度限制进行换行处理" wrapped_text_zh, text_height_zh = vd.wrap_text( text=test_text_zh, max_width=300, font=font_path, fontsize=30 ) print(wrapped_text_zh, text_height_zh) # verify chinese text is wrapped self.assertIn("\n", wrapped_text_zh) except Exception as e: self.fail(f"test wrap_text failed: {str(e)}") def test_rounded_subtitle_background_clip_has_transparent_corners(self): """ 圆角字幕背景只在用户显式开启时使用。这里直接验证生成的 RGBA 背景具备透明圆角和半透明中心，避免后续改动把圆角效果退化成实心矩形。 """ clip = vd._rounded_subtitle_background_clip( width=120, height=48, color="#123456", alpha=140, radius=16, ) try: frame = clip.get_frame(0) mask = clip.mask.get_frame(0) self.assertEqual(frame.shape[0:2], (48, 120)) self.assertEqual(tuple(frame[24, 60]), (18, 52, 86)) self.assertEqual(mask[0, 0], 0) self.assertGreater(mask[24, 60], 0.5) self.assertLess(mask[24, 60], 0.6) finally: clip.close() def test_get_temp_audio_dir_returns_system_temp_on_windows(self): with patch("sys.platform", "win32"): result = vd._get_temp_audio_dir("/some/output/dir") self.assertEqual(result, tempfile.gettempdir()) def test_get_temp_audio_dir_returns_output_dir_on_non_windows(self): for platform in ("linux", "darwin"): with self.subTest(platform=platform): with patch("sys.platform", platform): result = vd._get_temp_audio_dir("/some/output/dir") self.assertEqual(result, "/some/output/dir") class TestMaterialResolutionTolerance(unittest.TestCase): def test_accepts_material_at_the_nominal_minimum(self): self.assertTrue(vd.is_material_resolution_acceptable(480, 480)) def test_accepts_whatsapp_recompressed_portrait_clip(self): # WhatsApp delivers 9:16 clips as 478x850, two pixels under the # nominal 480 minimum. Rejecting them fails the whole task. self.assertTrue(vd.is_material_resolution_acceptable(478, 850)) def test_accepts_material_exactly_at_the_tolerance_bound(self): bound = vd._MIN_MATERIAL_DIMENSION - vd._MIN_DIMENSION_TOLERANCE self.assertTrue(vd.is_material_resolution_acceptable(bound, bound)) def test_rejects_material_just_below_the_tolerance_bound(self): bound = vd._MIN_MATERIAL_DIMENSION - vd._MIN_DIMENSION_TOLERANCE self.assertFalse(vd.is_material_resolution_acceptable(bound - 1, 850)) self.assertFalse(vd.is_material_resolution_acceptable(850, bound - 1)) def test_rejects_genuinely_low_resolution_material(self): self.assertFalse(vd.is_material_resolution_acceptable(320, 240)) if __name__ == "__main__": unittest.main()