Initial lip-sync service with command backend
This commit is contained in:
@@ -0,0 +1,22 @@
|
||||
# Tests
|
||||
|
||||
当前使用 Python 标准库 `unittest`,无需额外测试依赖。
|
||||
|
||||
```powershell
|
||||
uv run python -m unittest discover -s tests
|
||||
```
|
||||
|
||||
已覆盖:
|
||||
|
||||
1. `command` 后端模板渲染。
|
||||
2. 未知模板变量报错。
|
||||
3. 非零退出码的 stderr 尾部诊断与敏感字段脱敏。
|
||||
4. `command` 后端成功产出非空输出文件。
|
||||
|
||||
后续建议增加:
|
||||
|
||||
1. `normalize_audio_path` / `normalize_video_path` payload 解析测试。
|
||||
2. `copy_video` 或 `ffmpeg_mux` 集成测试。
|
||||
3. Gradio `/process_single` 兼容测试。
|
||||
|
||||
不要提交大音频/视频素材。测试素材可放在本机 `C:\test` 或通过环境变量指定。
|
||||
@@ -0,0 +1,110 @@
|
||||
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()
|
||||
Reference in New Issue
Block a user