lxbfYeaa/ContentGeneration
Template
1
0
Fork 0
forked from LiveCarta/ContentGeneration
Code Pull Requests Activity

Refactor pipeline to call script entrypoints directly

Browse Source
This commit is contained in:
Artsiom Siamashka
2026-04-03 16:15:19 +02:00
parent 74f8159eff
commit 008ee18ba8
8 changed files with 188 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files

98
tests/test_pipeline_full_process.py
View File

@@ -48,62 +48,72 @@ def test_full_generation_process_calls_all_scripts(monkeypatch) -> None:
log_level="DEBUG",
)
executed_scripts: list[str] = []
executed_steps: list[str] = []
expected_scripts = [
"generate_script.py",
"generate_audios.py",
"generate_images.py",
"generate_videos.py",
"merge_audio_video.py",
"concat_merged.py",
expected_steps = [
"generate_script",
"generate_audios",
"generate_images",
"generate_videos",
"merge_audio_video",
"concat_merged",
]
def fake_subprocess_run(cmd: list[str], check: bool, cwd: str | None = None):
script_name = Path(cmd[1]).name if len(cmd) > 1 else ""
if script_name not in expected_scripts:
pytest.fail(f"Unexpected external process call: {cmd}")
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
executed_scripts.append(script_name)
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
if script_name == "generate_script.py":
payload = {
"shots": [
{
"shot_number": 1,
"image_description": "A test image",
"voiceover": "A test voiceover",
}
]
}
reel_script.write_text(json.dumps(payload))
elif script_name == "generate_audios.py":
audios_dir.mkdir(parents=True, exist_ok=True)
(audios_dir / "output_1.mp3").write_bytes(b"audio")
elif script_name == "generate_images.py":
images_dir.mkdir(parents=True, exist_ok=True)
(images_dir / "shot_1.png").write_bytes(b"image")
elif script_name == "generate_videos.py":
videos_dir.mkdir(parents=True, exist_ok=True)
(videos_dir / "output_1.mp4").write_bytes(b"video")
elif script_name == "merge_audio_video.py":
merged_dir.mkdir(parents=True, exist_ok=True)
(merged_dir / "merged_1.mp4").write_bytes(b"merged")
elif script_name == "concat_merged.py":
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_bytes(b"final")
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
class Result:
returncode = 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
return Result()
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.subprocess, "run", fake_subprocess_run)
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_scripts == expected_scripts
assert executed_steps == expected_steps