初始化提交

test/services/test_video.py

@@ -0,0 +1,315 @@
+
+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.test_img_path = os.path.join(resources_dir, "1.png")
+    
+    def tearDown(self):
+        pass
+    
+    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(vd.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(
+            vd.shutil, "which", return_value=None
+        ), patch.dict(sys.modules, {"imageio_ffmpeg": fake_imageio_ffmpeg}):
+            self.assertEqual(vd.get_ffmpeg_binary(), "/tmp/bundled-ffmpeg")
+
+    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_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)}")
+
+if __name__ == "__main__":
+    unittest.main()