wehub-resource-sync
465 lines
17 KiB
Python
465 lines
17 KiB
Python
import io
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from contextlib import redirect_stdout
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
from uuid import uuid4
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
|
|
|
|
import cli
|
|
|
|
|
|
class TestCli(unittest.TestCase):
|
|
def test_default_voice_is_valid_edge_tts_voice(self):
|
|
args = cli.parse_args(["--video-subject", "测试主题"])
|
|
params = cli.build_video_params(args)
|
|
|
|
self.assertEqual(params.voice_name, "zh-CN-XiaoxiaoNeural-Female")
|
|
|
|
def test_complete_script_can_replace_video_subject(self):
|
|
args = cli.parse_args(["--video-script", "完整的视频文案"])
|
|
params = cli.build_video_params(args)
|
|
|
|
self.assertEqual(params.video_subject, "")
|
|
self.assertEqual(params.video_script, "完整的视频文案")
|
|
|
|
def test_subject_or_script_is_required(self):
|
|
with self.assertRaises(SystemExit) as cm:
|
|
cli.parse_args([])
|
|
|
|
self.assertEqual(cm.exception.code, 2)
|
|
|
|
def test_build_video_params_with_local_materials(self):
|
|
args = cli.parse_args(
|
|
[
|
|
"--video-subject",
|
|
"测试主题",
|
|
"--video-source",
|
|
"local",
|
|
"--video-materials",
|
|
"a.mp4, ,b.jpg,",
|
|
"--video-terms",
|
|
"foo, bar",
|
|
]
|
|
)
|
|
|
|
params = cli.build_video_params(args)
|
|
materials = params.video_materials
|
|
|
|
self.assertEqual(params.video_subject, "测试主题")
|
|
self.assertEqual(params.video_source, "local")
|
|
self.assertEqual([m.url for m in materials], ["a.mp4", "b.jpg"])
|
|
self.assertTrue(all(m.provider == "local" for m in materials))
|
|
self.assertEqual(params.video_terms, ["foo", "bar"])
|
|
|
|
def test_run_cli_dispatches_task_start(self):
|
|
with patch("app.services.task.start", return_value={"script": "ok"}) as start, patch(
|
|
"app.utils.utils.get_uuid", return_value="task-123"
|
|
), patch("builtins.print") as print_mock:
|
|
code = cli.run_cli(["--video-subject", "命令行测试", "--stop-at", "script"])
|
|
|
|
self.assertEqual(code, 0)
|
|
self.assertTrue(start.called)
|
|
kwargs = start.call_args.kwargs
|
|
self.assertEqual(kwargs["task_id"], "task-123")
|
|
self.assertEqual(kwargs["stop_at"], "script")
|
|
self.assertEqual(kwargs["params"].video_subject, "命令行测试")
|
|
print_mock.assert_called_once()
|
|
|
|
def test_run_cli_returns_error_when_task_fails(self):
|
|
with patch("app.services.task.start", return_value=None), patch(
|
|
"app.utils.utils.get_uuid", return_value="task-456"
|
|
), patch.object(cli.logger, "error") as log_error:
|
|
code = cli.run_cli(["--video-subject", "失败场景"])
|
|
|
|
self.assertEqual(code, 1)
|
|
log_error.assert_called_once()
|
|
|
|
def test_subtitle_enabled_by_default(self):
|
|
args = cli.parse_args(["--video-subject", "test"])
|
|
params = cli.build_video_params(args)
|
|
self.assertTrue(params.subtitle_enabled)
|
|
|
|
def test_subtitle_disabled_with_no_flag(self):
|
|
args = cli.parse_args(["--video-subject", "test", "--no-subtitle-enabled"])
|
|
params = cli.build_video_params(args)
|
|
self.assertFalse(params.subtitle_enabled)
|
|
|
|
def test_coverr_video_source_accepted(self):
|
|
args = cli.parse_args(["--video-subject", "test", "--video-source", "coverr"])
|
|
params = cli.build_video_params(args)
|
|
self.assertEqual(params.video_source, "coverr")
|
|
|
|
def test_build_video_params_with_script_video_and_audio_options(self):
|
|
args = cli.parse_args(
|
|
[
|
|
"--video-subject",
|
|
"test",
|
|
"--video-language",
|
|
"en",
|
|
"--paragraph-number",
|
|
"3",
|
|
"--video-script-prompt",
|
|
"use a lighter tone",
|
|
"--custom-system-prompt",
|
|
"write concise short-form scripts",
|
|
"--video-concat-mode",
|
|
"sequential",
|
|
"--video-transition-mode",
|
|
"fade-in",
|
|
"--video-clip-duration",
|
|
"4",
|
|
"--match-materials-to-script",
|
|
"--voice-volume",
|
|
"1.2",
|
|
"--voice-rate",
|
|
"1.1",
|
|
"--bgm-type",
|
|
"custom",
|
|
"--bgm-file",
|
|
"output001.mp3",
|
|
"--bgm-volume",
|
|
"0.3",
|
|
"--n-threads",
|
|
"4",
|
|
]
|
|
)
|
|
|
|
params = cli.build_video_params(args)
|
|
|
|
self.assertEqual(params.video_language, "en")
|
|
self.assertEqual(params.paragraph_number, 3)
|
|
self.assertEqual(params.video_script_prompt, "use a lighter tone")
|
|
self.assertEqual(params.custom_system_prompt, "write concise short-form scripts")
|
|
self.assertEqual(params.video_concat_mode, "sequential")
|
|
self.assertEqual(params.video_transition_mode, "FadeIn")
|
|
self.assertEqual(params.video_clip_duration, 4)
|
|
self.assertTrue(params.match_materials_to_script)
|
|
self.assertEqual(params.voice_volume, 1.2)
|
|
self.assertEqual(params.voice_rate, 1.1)
|
|
self.assertEqual(params.bgm_type, "custom")
|
|
self.assertEqual(params.bgm_file, "output001.mp3")
|
|
self.assertEqual(params.bgm_volume, 0.3)
|
|
self.assertEqual(params.n_threads, 4)
|
|
|
|
def test_custom_audio_file_maps_to_video_params(self):
|
|
args = cli.parse_args(
|
|
[
|
|
"--video-subject",
|
|
"test",
|
|
"--custom-audio-file",
|
|
"voiceover.mp3",
|
|
]
|
|
)
|
|
params = cli.build_video_params(args)
|
|
|
|
self.assertEqual(params.custom_audio_file, "voiceover.mp3")
|
|
|
|
def test_build_video_params_with_subtitle_style_options(self):
|
|
args = cli.parse_args(
|
|
[
|
|
"--video-subject",
|
|
"test",
|
|
"--font-name",
|
|
"MicrosoftYaHeiBold.ttc",
|
|
"--subtitle-position",
|
|
"custom",
|
|
"--custom-position",
|
|
"42.5",
|
|
"--text-fore-color",
|
|
"#AABBCC",
|
|
"--font-size",
|
|
"72",
|
|
"--stroke-color",
|
|
"#112233",
|
|
"--stroke-width",
|
|
"2.5",
|
|
"--subtitle-background-color",
|
|
"#000001",
|
|
"--rounded-subtitle-background",
|
|
]
|
|
)
|
|
|
|
params = cli.build_video_params(args)
|
|
|
|
self.assertEqual(params.font_name, "MicrosoftYaHeiBold.ttc")
|
|
self.assertEqual(params.subtitle_position, "custom")
|
|
self.assertEqual(params.custom_position, 42.5)
|
|
self.assertEqual(params.text_fore_color, "#AABBCC")
|
|
self.assertEqual(params.font_size, 72)
|
|
self.assertEqual(params.stroke_color, "#112233")
|
|
self.assertEqual(params.stroke_width, 2.5)
|
|
self.assertEqual(params.text_background_color, "#000001")
|
|
self.assertTrue(params.rounded_subtitle_background)
|
|
|
|
def test_disabled_subtitle_background_rejects_rounding(self):
|
|
with self.assertRaises(SystemExit) as cm:
|
|
cli.parse_args(
|
|
[
|
|
"--video-subject",
|
|
"test",
|
|
"--no-subtitle-background-enabled",
|
|
"--rounded-subtitle-background",
|
|
]
|
|
)
|
|
|
|
self.assertEqual(cm.exception.code, 2)
|
|
|
|
def test_bgm_type_none_maps_to_disabled_background_music(self):
|
|
args = cli.parse_args(["--video-subject", "test", "--bgm-type", "none"])
|
|
params = cli.build_video_params(args)
|
|
self.assertEqual(params.bgm_type, "")
|
|
|
|
def test_local_material_filename_resolved_to_absolute_path(self):
|
|
"""After preprocess_video, material.url should be an absolute path, not a bare filename."""
|
|
import os
|
|
import tempfile
|
|
from app.utils import utils
|
|
from app.services import video as vd
|
|
from app.models.schema import MaterialInfo
|
|
|
|
local_videos_dir = utils.storage_dir("local_videos", create=True)
|
|
# Create a minimal valid video file for testing
|
|
test_filename = "_cli_test_resolve.mp4"
|
|
test_filepath = os.path.join(local_videos_dir, test_filename)
|
|
# We need a real video file; use a tiny one via moviepy
|
|
try:
|
|
from moviepy import ColorClip
|
|
clip = ColorClip(size=(640, 640), color=(0, 0, 0), duration=1)
|
|
clip.write_videofile(test_filepath, fps=1, logger=None)
|
|
clip.close()
|
|
except Exception:
|
|
self.skipTest("moviepy not available for creating test video")
|
|
|
|
try:
|
|
materials = [MaterialInfo(provider="local", url=test_filename, duration=0)]
|
|
result = vd.preprocess_video(materials=materials, clip_duration=4)
|
|
self.assertTrue(len(result) > 0, "preprocess_video should return valid materials")
|
|
self.assertTrue(
|
|
os.path.isabs(result[0].url),
|
|
f"material url should be absolute path, got: {result[0].url}",
|
|
)
|
|
self.assertEqual(result[0].url, test_filepath)
|
|
finally:
|
|
if os.path.exists(test_filepath):
|
|
os.remove(test_filepath)
|
|
|
|
|
|
def test_local_source_requires_video_materials(self):
|
|
with self.assertRaises(SystemExit) as cm:
|
|
cli.parse_args(["--video-subject", "test", "--video-source", "local"])
|
|
self.assertNotEqual(cm.exception.code, 0)
|
|
|
|
def test_local_source_does_not_require_materials_before_material_stage(self):
|
|
for stop_at in ("script", "audio", "subtitle"):
|
|
with self.subTest(stop_at=stop_at):
|
|
args = cli.parse_args(
|
|
[
|
|
"--video-subject",
|
|
"test",
|
|
"--video-source",
|
|
"local",
|
|
"--stop-at",
|
|
stop_at,
|
|
]
|
|
)
|
|
self.assertEqual(args.stop_at, stop_at)
|
|
|
|
def test_local_source_stop_at_terms_rejected(self):
|
|
with self.assertRaises(SystemExit) as cm:
|
|
cli.parse_args([
|
|
"--video-subject", "test",
|
|
"--video-source", "local",
|
|
"--video-materials", "a.mp4",
|
|
"--stop-at", "terms",
|
|
])
|
|
self.assertNotEqual(cm.exception.code, 0)
|
|
|
|
def test_video_materials_rejected_for_online_source(self):
|
|
with self.assertRaises(SystemExit) as cm:
|
|
cli.parse_args(
|
|
[
|
|
"--video-subject",
|
|
"test",
|
|
"--video-source",
|
|
"pexels",
|
|
"--video-materials",
|
|
"a.mp4",
|
|
]
|
|
)
|
|
|
|
self.assertEqual(cm.exception.code, 2)
|
|
|
|
def test_custom_bgm_requires_file_and_file_implies_custom_mode(self):
|
|
with self.assertRaises(SystemExit):
|
|
cli.parse_args(
|
|
["--video-subject", "test", "--bgm-type", "custom"]
|
|
)
|
|
|
|
args = cli.parse_args(
|
|
["--video-subject", "test", "--bgm-file", "output001.mp3"]
|
|
)
|
|
self.assertEqual(args.bgm_type, "custom")
|
|
|
|
def test_invalid_aspect_and_non_finite_numbers_are_argument_errors(self):
|
|
invalid_argvs = [
|
|
["--video-subject", "test", "--video-aspect", "invalid"],
|
|
["--video-subject", "test", "--custom-position", "nan"],
|
|
["--video-subject", "test", "--voice-rate", "inf"],
|
|
]
|
|
for argv in invalid_argvs:
|
|
with self.subTest(argv=argv), self.assertRaises(SystemExit) as cm:
|
|
cli.parse_args(argv)
|
|
self.assertEqual(cm.exception.code, 2)
|
|
|
|
def test_custom_position_requires_custom_subtitle_position(self):
|
|
with self.assertRaises(SystemExit) as cm:
|
|
cli.parse_args(
|
|
["--video-subject", "test", "--custom-position", "50"]
|
|
)
|
|
|
|
self.assertEqual(cm.exception.code, 2)
|
|
|
|
def test_task_id_must_be_uuid(self):
|
|
task_id = str(uuid4())
|
|
args = cli.parse_args(
|
|
["--video-subject", "test", "--task-id", task_id]
|
|
)
|
|
self.assertEqual(args.task_id, task_id)
|
|
|
|
with self.assertRaises(SystemExit) as cm:
|
|
cli.parse_args(
|
|
["--video-subject", "test", "--task-id", "../../escape"]
|
|
)
|
|
self.assertEqual(cm.exception.code, 2)
|
|
|
|
def test_prepare_cli_files_accepts_relative_and_absolute_materials(self):
|
|
with (
|
|
tempfile.TemporaryDirectory() as source_dir,
|
|
tempfile.TemporaryDirectory() as managed_dir,
|
|
):
|
|
relative_file = Path(source_dir) / "relative.mp4"
|
|
absolute_file = Path(source_dir) / "absolute.jpg"
|
|
relative_file.write_bytes(b"relative-video")
|
|
absolute_file.write_bytes(b"absolute-image")
|
|
old_cwd = os.getcwd()
|
|
try:
|
|
os.chdir(source_dir)
|
|
args = cli.parse_args(
|
|
[
|
|
"--video-subject",
|
|
"test",
|
|
"--video-source",
|
|
"local",
|
|
"--video-materials",
|
|
f"relative.mp4,{absolute_file}",
|
|
]
|
|
)
|
|
params = cli.build_video_params(args)
|
|
with patch("app.utils.utils.storage_dir", return_value=managed_dir):
|
|
cli.prepare_cli_files(params, stop_at="video")
|
|
finally:
|
|
os.chdir(old_cwd)
|
|
|
|
prepared_paths = [Path(item.url) for item in params.video_materials]
|
|
self.assertTrue(all(path.parent == Path(managed_dir) for path in prepared_paths))
|
|
self.assertEqual(
|
|
{path.read_bytes() for path in prepared_paths},
|
|
{b"relative-video", b"absolute-image"},
|
|
)
|
|
|
|
def test_prepare_cli_files_rejects_missing_material_before_task_start(self):
|
|
args = cli.parse_args(
|
|
[
|
|
"--video-subject",
|
|
"test",
|
|
"--video-source",
|
|
"local",
|
|
"--video-materials",
|
|
"missing.mp4",
|
|
]
|
|
)
|
|
params = cli.build_video_params(args)
|
|
|
|
with tempfile.TemporaryDirectory() as managed_dir, patch(
|
|
"app.utils.utils.storage_dir", return_value=managed_dir
|
|
):
|
|
with self.assertRaisesRegex(ValueError, "does not exist"):
|
|
cli.prepare_cli_files(params, stop_at="video")
|
|
|
|
def test_run_cli_rejects_missing_material_before_starting_task(self):
|
|
with patch("app.services.task.start") as start:
|
|
code = cli.run_cli(
|
|
[
|
|
"--video-subject",
|
|
"test",
|
|
"--video-source",
|
|
"local",
|
|
"--video-materials",
|
|
"missing.mp4",
|
|
]
|
|
)
|
|
|
|
self.assertEqual(code, 2)
|
|
start.assert_not_called()
|
|
|
|
def test_prepare_cli_files_resolves_relative_custom_audio(self):
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
audio_file = Path(temp_dir) / "voiceover.mp3"
|
|
audio_file.write_bytes(b"audio")
|
|
old_cwd = os.getcwd()
|
|
try:
|
|
os.chdir(temp_dir)
|
|
args = cli.parse_args(
|
|
[
|
|
"--video-subject",
|
|
"test",
|
|
"--custom-audio-file",
|
|
"voiceover.mp3",
|
|
"--stop-at",
|
|
"audio",
|
|
]
|
|
)
|
|
params = cli.build_video_params(args)
|
|
cli.prepare_cli_files(params, stop_at="audio")
|
|
finally:
|
|
os.chdir(old_cwd)
|
|
|
|
self.assertEqual(params.custom_audio_file, str(audio_file.resolve()))
|
|
|
|
def test_help_documents_defaults_paths_stages_and_exit_codes(self):
|
|
output = io.StringIO()
|
|
with redirect_stdout(output), self.assertRaises(SystemExit) as cm:
|
|
cli.parse_args(["--help"])
|
|
|
|
self.assertEqual(cm.exception.code, 0)
|
|
help_text = output.getvalue()
|
|
self.assertIn("zh-CN-XiaoxiaoNeural-Female", help_text)
|
|
self.assertIn("current working directory", help_text)
|
|
self.assertIn("Pipeline stages:", help_text)
|
|
self.assertIn("exit with 2", help_text)
|
|
|
|
def test_help_does_not_initialize_application_or_write_logs(self):
|
|
"""帮助命令应独立于业务配置加载，便于用户查看和脚本采集。"""
|
|
project_root = Path(__file__).parent.parent.parent
|
|
result = subprocess.run(
|
|
[sys.executable, str(project_root / "cli.py"), "--help"],
|
|
cwd=project_root,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 0)
|
|
self.assertIn("Generate MoneyPrinterTurbo videos", result.stdout)
|
|
self.assertEqual(result.stderr, "")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|