cat-shark
111 lines
4.0 KiB
Python
111 lines
4.0 KiB
Python
from __future__ import annotations
|
|
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from digital_human_lipsync_service.backends import (
|
|
LipSyncError,
|
|
quote_for_shell,
|
|
render_command_template,
|
|
run_command_backend,
|
|
)
|
|
from digital_human_lipsync_service.config import Settings
|
|
|
|
|
|
def make_settings(tmp_path: Path, command_template: str) -> Settings:
|
|
output_dir = tmp_path / "outputs"
|
|
work_dir = tmp_path / "work"
|
|
output_dir.mkdir()
|
|
work_dir.mkdir()
|
|
return Settings(
|
|
host="127.0.0.1",
|
|
port=7860,
|
|
share=False,
|
|
backend="command",
|
|
ffmpeg="ffmpeg",
|
|
output_dir=output_dir,
|
|
work_dir=work_dir,
|
|
command_cwd=work_dir,
|
|
command_template=command_template,
|
|
command_timeout_seconds=10,
|
|
concurrency_limit=1,
|
|
)
|
|
|
|
|
|
class CommandBackendTests(unittest.TestCase):
|
|
def test_render_command_template_supports_raw_and_quoted_paths(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
settings = make_settings(root, "model --audio {audio} --face {video_q} --out {output_q} --cwd {cwd}")
|
|
audio = root / "input audio.wav"
|
|
video = root / "input video.mp4"
|
|
output = root / "out video.mp4"
|
|
|
|
command = render_command_template(settings, audio, video, output)
|
|
|
|
self.assertIn(str(audio), command)
|
|
self.assertIn("input video.mp4", command)
|
|
self.assertIn("out video.mp4", command)
|
|
self.assertIn(str(settings.command_cwd), command)
|
|
|
|
def test_render_command_template_rejects_unknown_variable(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
settings = make_settings(root, "model --bad {missing}")
|
|
|
|
with self.assertRaisesRegex(LipSyncError, "Unknown LIPSYNC_COMMAND_TEMPLATE variable"):
|
|
render_command_template(settings, root / "a.wav", root / "v.mp4", root / "o.mp4")
|
|
|
|
def test_run_command_backend_reports_exit_code_and_stderr_tail(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
script = root / "fail_backend.py"
|
|
script.write_text(
|
|
"import sys\n"
|
|
"sys.stderr.write('api_key=super-secret\\nmodel failed near frame 12\\n')\n"
|
|
"raise SystemExit(7)\n",
|
|
encoding="utf-8",
|
|
)
|
|
settings = make_settings(root, f"{quote_for_shell(sys.executable)} {quote_for_shell(script)}")
|
|
audio = root / "a.wav"
|
|
video = root / "v.mp4"
|
|
output = root / "out.mp4"
|
|
audio.write_bytes(b"audio")
|
|
video.write_bytes(b"video")
|
|
|
|
with self.assertRaises(LipSyncError) as context:
|
|
run_command_backend(settings, audio, video, output)
|
|
|
|
message = str(context.exception)
|
|
self.assertIn("exit_code=7", message)
|
|
self.assertIn("model failed near frame 12", message)
|
|
self.assertIn("api_key=***", message)
|
|
self.assertNotIn("super-secret", message)
|
|
|
|
def test_run_command_backend_accepts_non_empty_output_file(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
script = root / "ok_backend.py"
|
|
script.write_text(
|
|
"import pathlib, sys\n"
|
|
"pathlib.Path(sys.argv[1]).write_bytes(b'fake mp4')\n",
|
|
encoding="utf-8",
|
|
)
|
|
settings = make_settings(root, f"{quote_for_shell(sys.executable)} {quote_for_shell(script)} {{output_q}}")
|
|
audio = root / "a.wav"
|
|
video = root / "v.mp4"
|
|
output = root / "out.mp4"
|
|
audio.write_bytes(b"audio")
|
|
video.write_bytes(b"video")
|
|
|
|
result = run_command_backend(settings, audio, video, output)
|
|
|
|
self.assertEqual(output, result)
|
|
self.assertGreater(output.stat().st_size, 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|