wehub-resource-sync
711 lines
28 KiB
Python
711 lines
28 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.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()
|