316 lines
12 KiB
Python
316 lines
12 KiB
Python
|
||
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()
|