forked from LiveCarta/ContentGeneration
120 lines
4.2 KiB
Python
120 lines
4.2 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
import tempfile
|
|
from argparse import Namespace
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
# Avoid requiring boto3 for orchestration tests.
|
|
if "boto3" not in sys.modules:
|
|
sys.modules["boto3"] = SimpleNamespace(client=lambda *args, **kwargs: object())
|
|
|
|
import run_video_pipeline as pipeline
|
|
|
|
|
|
def test_full_generation_process_calls_all_scripts(monkeypatch) -> None:
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
base_dir = Path(tmpdir)
|
|
hunyuan_dir = base_dir / "HunyuanVideo-1.5"
|
|
images_dir = base_dir / "images"
|
|
videos_dir = base_dir / "videos"
|
|
audios_dir = base_dir / "audios"
|
|
merged_dir = base_dir / "merged"
|
|
output_path = base_dir / "results" / "final_output.mp4"
|
|
reel_script = base_dir / "reel_script.json"
|
|
|
|
hunyuan_dir.mkdir(parents=True)
|
|
(base_dir / "topic_description.txt").write_text("Test topic")
|
|
|
|
args = Namespace(
|
|
base_dir=base_dir,
|
|
hunyuan_dir=hunyuan_dir,
|
|
reel_script=reel_script,
|
|
images_dir=images_dir,
|
|
videos_dir=videos_dir,
|
|
audios_dir=audios_dir,
|
|
merged_dir=merged_dir,
|
|
output=output_path,
|
|
seed=1,
|
|
skip_generate=False,
|
|
skip_audio_generate=False,
|
|
skip_merge=False,
|
|
skip_concat=False,
|
|
skip_s3_upload=True,
|
|
log_level="DEBUG",
|
|
)
|
|
|
|
executed_steps: list[str] = []
|
|
|
|
expected_steps = [
|
|
"generate_script",
|
|
"generate_audios",
|
|
"generate_images",
|
|
"generate_videos",
|
|
"merge_audio_video",
|
|
"concat_merged",
|
|
]
|
|
|
|
def fake_generate_script_main(argv=None) -> int:
|
|
executed_steps.append("generate_script")
|
|
payload = {
|
|
"shots": [
|
|
{
|
|
"shot_number": 1,
|
|
"image_description": "A test image",
|
|
"voiceover": "A test voiceover",
|
|
}
|
|
]
|
|
}
|
|
reel_script.write_text(json.dumps(payload))
|
|
return 0
|
|
|
|
def fake_generate_audios_main(argv=None) -> int:
|
|
executed_steps.append("generate_audios")
|
|
audios_dir.mkdir(parents=True, exist_ok=True)
|
|
(audios_dir / "output_1.mp3").write_bytes(b"audio")
|
|
return 0
|
|
|
|
def fake_generate_images_main(argv=None) -> int:
|
|
executed_steps.append("generate_images")
|
|
images_dir.mkdir(parents=True, exist_ok=True)
|
|
(images_dir / "shot_1.png").write_bytes(b"image")
|
|
return 0
|
|
|
|
def fake_generate_videos_main(argv=None) -> int:
|
|
executed_steps.append("generate_videos")
|
|
videos_dir.mkdir(parents=True, exist_ok=True)
|
|
(videos_dir / "output_1.mp4").write_bytes(b"video")
|
|
return 0
|
|
|
|
def fake_merge_audio_video_main(argv=None) -> int:
|
|
executed_steps.append("merge_audio_video")
|
|
merged_dir.mkdir(parents=True, exist_ok=True)
|
|
(merged_dir / "merged_1.mp4").write_bytes(b"merged")
|
|
return 0
|
|
|
|
def fake_concat_merged_main(argv=None) -> int:
|
|
executed_steps.append("concat_merged")
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
|
output_path.write_bytes(b"final")
|
|
return 0
|
|
|
|
monkeypatch.setattr(pipeline, "parse_args", lambda: args)
|
|
monkeypatch.setattr(pipeline.generate_script, "main", fake_generate_script_main)
|
|
monkeypatch.setattr(pipeline.generate_audios, "main", fake_generate_audios_main)
|
|
monkeypatch.setattr(pipeline.generate_images, "main", fake_generate_images_main)
|
|
monkeypatch.setattr(pipeline.generate_videos, "main", fake_generate_videos_main)
|
|
monkeypatch.setattr(pipeline.merge_audio_video, "main", fake_merge_audio_video_main)
|
|
monkeypatch.setattr(pipeline.concat_merged, "main", fake_concat_merged_main)
|
|
|
|
rc = pipeline.main()
|
|
|
|
assert rc == 0
|
|
assert output_path.exists()
|
|
# Coverage check for orchestration: ensure every required script stage was called.
|
|
assert executed_steps == expected_steps
|