|
|
@@ -0,0 +1,333 @@
|
|
|
+import io
|
|
|
+import json
|
|
|
+import subprocess
|
|
|
+import sys
|
|
|
+import tempfile
|
|
|
+import unittest
|
|
|
+from contextlib import redirect_stderr, redirect_stdout
|
|
|
+from pathlib import Path
|
|
|
+from unittest.mock import patch
|
|
|
+
|
|
|
+import fixtures
|
|
|
+
|
|
|
+
|
|
|
+class NarrationPublicationContract(unittest.TestCase):
|
|
|
+ def setUp(self):
|
|
|
+ self.module = fixtures.load_script("narrate")
|
|
|
+ self.directory = tempfile.TemporaryDirectory()
|
|
|
+ self.addCleanup(self.directory.cleanup)
|
|
|
+ self.root = Path(self.directory.name)
|
|
|
+ self.scenes = self.root / "scenes.yaml"
|
|
|
+ self.output = self.root / "narration"
|
|
|
+
|
|
|
+ def run_narrate(self, scenes, synthesize, *, verify="off", expected=0):
|
|
|
+ self.scenes.write_text(json.dumps({"scenes": scenes}), encoding="utf-8")
|
|
|
+ argv = ["narrate", str(self.scenes), str(self.output), "--engine", "piper",
|
|
|
+ "--verify", verify]
|
|
|
+ stderr = io.StringIO()
|
|
|
+ with patch.object(sys, "argv", argv), \
|
|
|
+ patch.object(self.module.shutil, "which", return_value="ffprobe"), \
|
|
|
+ patch.object(self.module, "openai_key", return_value=None), \
|
|
|
+ patch.object(self.module, "say_piper", side_effect=synthesize), \
|
|
|
+ patch.object(self.module, "duration", return_value=1.25), \
|
|
|
+ patch.object(self.module, "transcribe_local", return_value=None), \
|
|
|
+ redirect_stdout(io.StringIO()), redirect_stderr(stderr):
|
|
|
+ self.assertEqual(self.module.main(), expected, stderr.getvalue())
|
|
|
+ manifest = self.output / "manifest.json"
|
|
|
+ return json.loads(manifest.read_text(encoding="utf-8")) if manifest.exists() else []
|
|
|
+
|
|
|
+ def test_acceptance_is_withdrawn_before_replacing_accepted_bytes(self):
|
|
|
+ old_wav = self.output / "first.wav"
|
|
|
+ self.output.mkdir()
|
|
|
+ old_wav.write_bytes(b"accepted bytes")
|
|
|
+ synthesis = {"engine": "piper", "voice": self.module.PIPER_VOICE,
|
|
|
+ "model": self.module.PIPER_VOICE}
|
|
|
+ (self.output / "manifest.json").write_text(json.dumps([{
|
|
|
+ "id": "first", "text": "Old words", "wav": "first.wav",
|
|
|
+ "duration": 1.0, "synthesis": synthesis,
|
|
|
+ }]), encoding="utf-8")
|
|
|
+
|
|
|
+ def synthesize(text, wav, voice):
|
|
|
+ published = json.loads((self.output / "manifest.json").read_text())
|
|
|
+ self.assertNotIn("first", [entry["id"] for entry in published])
|
|
|
+ wav.write_bytes(b"replacement bytes")
|
|
|
+
|
|
|
+ manifest = self.run_narrate(
|
|
|
+ [{"id": "first", "narration": "New words"}], synthesize
|
|
|
+ )
|
|
|
+
|
|
|
+ self.assertEqual(manifest[0]["text"], "New words")
|
|
|
+ self.assertEqual(old_wav.read_bytes(), b"replacement bytes")
|
|
|
+
|
|
|
+ def test_rejected_and_exceptional_takes_are_not_accepted_on_a_rerun(self):
|
|
|
+ attempts = []
|
|
|
+
|
|
|
+ def synthesize(text, wav, voice):
|
|
|
+ attempts.append((text, wav))
|
|
|
+ wav.write_bytes(f"{text}:{len(attempts)}".encode())
|
|
|
+ if text == "Reject this":
|
|
|
+ return "invented words that do not match this script at all"
|
|
|
+ if text == "Raise here":
|
|
|
+ raise RuntimeError("synthesis interrupted")
|
|
|
+
|
|
|
+ scenes = [{"id": "reject", "narration": "Reject this"},
|
|
|
+ {"id": "raise", "narration": "Raise here"}]
|
|
|
+ first = self.run_narrate(scenes, synthesize, expected=1)
|
|
|
+ second = self.run_narrate(scenes, synthesize, expected=1)
|
|
|
+
|
|
|
+ self.assertEqual(first, [])
|
|
|
+ self.assertEqual(second, [])
|
|
|
+ rejected_paths = [path for text, path in attempts if text == "Reject this"]
|
|
|
+ self.assertEqual(len({path.name for path in rejected_paths}), len(rejected_paths))
|
|
|
+ self.assertTrue(all(path.exists() for path in rejected_paths))
|
|
|
+
|
|
|
+ def test_duration_failure_leaves_only_prior_accepted_scenes_published(self):
|
|
|
+ scenes = [{"id": "accepted", "narration": "Accepted words"},
|
|
|
+ {"id": "broken", "narration": "Broken words"}]
|
|
|
+
|
|
|
+ def synthesize(text, wav, voice):
|
|
|
+ wav.write_bytes(text.encode())
|
|
|
+
|
|
|
+ self.scenes.write_text(json.dumps({"scenes": scenes}), encoding="utf-8")
|
|
|
+ argv = ["narrate", str(self.scenes), str(self.output), "--engine", "piper",
|
|
|
+ "--verify", "off"]
|
|
|
+ durations = iter((1.0, RuntimeError("ffprobe failed")))
|
|
|
+ with patch.object(sys, "argv", argv), \
|
|
|
+ patch.object(self.module.shutil, "which", return_value="ffprobe"), \
|
|
|
+ patch.object(self.module, "openai_key", return_value=None), \
|
|
|
+ patch.object(self.module, "say_piper", side_effect=synthesize), \
|
|
|
+ patch.object(self.module, "duration", side_effect=durations), \
|
|
|
+ redirect_stdout(io.StringIO()), redirect_stderr(io.StringIO()):
|
|
|
+ self.assertEqual(self.module.main(), 1)
|
|
|
+
|
|
|
+ manifest = json.loads((self.output / "manifest.json").read_text())
|
|
|
+ self.assertEqual([entry["id"] for entry in manifest], ["accepted"])
|
|
|
+
|
|
|
+ def test_two_accepted_scenes_are_published_together(self):
|
|
|
+ def synthesize(text, wav, voice):
|
|
|
+ wav.write_bytes(text.encode())
|
|
|
+
|
|
|
+ manifest = self.run_narrate([
|
|
|
+ {"id": "first", "narration": "First accepted scene"},
|
|
|
+ {"id": "second", "narration": "Second accepted scene"},
|
|
|
+ ], synthesize)
|
|
|
+ self.assertEqual([entry["id"] for entry in manifest], ["first", "second"])
|
|
|
+
|
|
|
+ def test_strict_unavailable_verification_withdraws_cached_acceptance(self):
|
|
|
+ def synthesize(text, wav, voice):
|
|
|
+ wav.write_bytes(b"accepted")
|
|
|
+
|
|
|
+ self.run_narrate([{"id": "clip", "narration": "Read these words"}], synthesize)
|
|
|
+ manifest = self.run_narrate(
|
|
|
+ [{"id": "clip", "narration": "Read these words"}], synthesize,
|
|
|
+ verify="on", expected=1,
|
|
|
+ )
|
|
|
+ self.assertEqual(manifest, [])
|
|
|
+
|
|
|
+ def test_unsupported_asr_comparison_obeys_auto_on_and_off_modes(self):
|
|
|
+ def synthesize(text, wav, voice):
|
|
|
+ wav.write_bytes(b"clip")
|
|
|
+
|
|
|
+ scenes = [{"id": "clip", "narration": "\u77ed\u6587"}]
|
|
|
+ for mode, expected in (("auto", 0), ("on", 1), ("off", 0)):
|
|
|
+ with self.subTest(mode=mode):
|
|
|
+ manifest = self.run_narrate(scenes, synthesize, verify=mode,
|
|
|
+ expected=expected)
|
|
|
+ self.assertEqual(bool(manifest), expected == 0)
|
|
|
+
|
|
|
+ def test_missing_ffprobe_stops_before_synthesis(self):
|
|
|
+ self.scenes.write_text(json.dumps({"scenes": [{"id": "clip", "narration": "Words"}]}),
|
|
|
+ encoding="utf-8")
|
|
|
+ with patch.object(sys, "argv", ["narrate", str(self.scenes), str(self.output)]), \
|
|
|
+ patch.object(self.module.shutil, "which", return_value=None), \
|
|
|
+ patch.object(self.module, "say_piper", side_effect=AssertionError("synthesized")), \
|
|
|
+ redirect_stderr(io.StringIO()):
|
|
|
+ with self.assertRaises(SystemExit):
|
|
|
+ self.module.main()
|
|
|
+
|
|
|
+ def test_cached_unsupported_chat_transcript_withdraws_acceptance_without_asr(self):
|
|
|
+ self.output.mkdir()
|
|
|
+ (self.output / "clip.wav").write_bytes(b"cached bytes")
|
|
|
+ synthesis = {"engine": "openai-chat", "voice": "nova",
|
|
|
+ "model": self.module.OPENAI_CHAT_MODEL}
|
|
|
+ (self.output / "manifest.json").write_text(json.dumps([{
|
|
|
+ "id": "clip", "text": "\u77ed\u6587", "wav": "clip.wav", "duration": 1.0,
|
|
|
+ "synthesis": synthesis,
|
|
|
+ }]), encoding="utf-8")
|
|
|
+ self.scenes.write_text(json.dumps({"scenes": [
|
|
|
+ {"id": "clip", "narration": "\u77ed\u6587"}
|
|
|
+ ]}), encoding="utf-8")
|
|
|
+ argv = ["narrate", str(self.scenes), str(self.output), "--engine", "openai-chat",
|
|
|
+ "--verify", "off"]
|
|
|
+ with patch.object(sys, "argv", argv), \
|
|
|
+ patch.object(self.module.shutil, "which", return_value="ffprobe"), \
|
|
|
+ patch.object(self.module, "openai_key", return_value="key"), \
|
|
|
+ patch.object(self.module, "say_openai_chat", side_effect=AssertionError("cached")), \
|
|
|
+ patch.object(self.module, "transcribe_local", side_effect=AssertionError("ASR")), \
|
|
|
+ redirect_stdout(io.StringIO()), redirect_stderr(io.StringIO()):
|
|
|
+ self.assertEqual(self.module.main(), 1)
|
|
|
+ self.assertEqual(json.loads((self.output / "manifest.json").read_text()), [])
|
|
|
+
|
|
|
+
|
|
|
+class NarrationComparisonContract(unittest.TestCase):
|
|
|
+ def setUp(self):
|
|
|
+ self.module = fixtures.load_script("narrate")
|
|
|
+
|
|
|
+ def test_comparison_marks_scripts_requiring_segmentation_or_no_words_unavailable(self):
|
|
|
+ for script, heard in (
|
|
|
+ ("你好世界", "こんにちは世界"),
|
|
|
+ ("短文", "短文"),
|
|
|
+ ("mixed 日本語 words", "mixed 日本語 words"),
|
|
|
+ ("*** !!!", "*** !!!"),
|
|
|
+ ):
|
|
|
+ with self.subTest(script=script):
|
|
|
+ self.assertIsNone(self.module.structural_drift(script, heard))
|
|
|
+
|
|
|
+ def test_comparison_accepts_multiline_accented_latin_and_spaced_cyrillic(self):
|
|
|
+ for script, heard in (
|
|
|
+ ("Caf\u00e9\nna\u00efve", "CAF\u00c9 na\u00efve"),
|
|
|
+ ("\u041f\u0440\u0438\u0432\u0435\u0442 \u043c\u0438\u0440", "\u043f\u0440\u0438\u0432\u0435\u0442 \u043c\u0438\u0440"),
|
|
|
+ ("\uc548\ub155 \uc138\uacc4", "\uc548\ub155 \uc138\uacc4"),
|
|
|
+ ):
|
|
|
+ with self.subTest(script=script):
|
|
|
+ self.assertEqual(self.module.structural_drift(script, heard), (0.0, 0))
|
|
|
+
|
|
|
+ def test_drift_check_reports_unavailable_comparison_as_nonzero(self):
|
|
|
+ with tempfile.TemporaryDirectory() as directory:
|
|
|
+ root = Path(directory)
|
|
|
+ script = root / "script.txt"
|
|
|
+ heard = root / "heard.txt"
|
|
|
+ script.write_text("\u77ed\u6587", encoding="utf-8")
|
|
|
+ heard.write_text("\u77ed\u6587", encoding="utf-8")
|
|
|
+ output = io.StringIO()
|
|
|
+ with patch.object(sys, "argv", ["narrate", "--drift-check", str(script), str(heard)]), \
|
|
|
+ redirect_stdout(output):
|
|
|
+ self.assertEqual(self.module.main(), 1)
|
|
|
+ self.assertIn("comparison unavailable", output.getvalue())
|
|
|
+
|
|
|
+
|
|
|
+class AssemblyNarrationContract(unittest.TestCase):
|
|
|
+ def setUp(self):
|
|
|
+ self.module = fixtures.load_script("assemble")
|
|
|
+ self.directory = tempfile.TemporaryDirectory()
|
|
|
+ self.addCleanup(self.directory.cleanup)
|
|
|
+ self.root = Path(self.directory.name)
|
|
|
+ self.scenes = self.root / "scenes.yaml"
|
|
|
+ self.narration = self.root / "narration"
|
|
|
+ self.narration.mkdir()
|
|
|
+
|
|
|
+ def assemble(self, scenes, *, manifest=None, run_side_effect=None):
|
|
|
+ self.scenes.write_text(json.dumps({"scenes": scenes}), encoding="utf-8")
|
|
|
+ if manifest is not None:
|
|
|
+ (self.narration / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
|
|
|
+ calls = []
|
|
|
+
|
|
|
+ def run(command, **kwargs):
|
|
|
+ calls.append(command)
|
|
|
+ if run_side_effect is not None:
|
|
|
+ return run_side_effect(command)
|
|
|
+ return subprocess.CompletedProcess(command, 0, "1.0", "")
|
|
|
+
|
|
|
+ argv = ["assemble", str(self.scenes), str(self.root / "out.mp4"),
|
|
|
+ "--narration", str(self.narration), "--work", str(self.root / "work")]
|
|
|
+ with patch.object(sys, "argv", argv), \
|
|
|
+ patch.object(self.module.shutil, "which", return_value="tool"), \
|
|
|
+ patch.object(self.module, "find_browser", return_value=None), \
|
|
|
+ patch.object(self.module, "run", side_effect=run), \
|
|
|
+ redirect_stdout(io.StringIO()), redirect_stderr(io.StringIO()):
|
|
|
+ return self.module.main(), calls
|
|
|
+
|
|
|
+ def test_required_narration_contract_fails_before_encoding(self):
|
|
|
+ with self.assertRaises(SystemExit):
|
|
|
+ self.assemble([{"id": "spoken", "kind": "image", "src": "still.png",
|
|
|
+ "narration": "Expected words"}],
|
|
|
+ run_side_effect=lambda command: (_ for _ in ()).throw(
|
|
|
+ AssertionError("encoding started")
|
|
|
+ ))
|
|
|
+
|
|
|
+ def test_missing_entry_wav_or_matching_text_fails_before_encoding(self):
|
|
|
+ (self.root / "still.png").write_bytes(b"image sentinel")
|
|
|
+ cases = (
|
|
|
+ ([], "missing entry"),
|
|
|
+ ([{"id": "spoken", "text": "Expected words", "wav": "missing.wav"}], "missing WAV"),
|
|
|
+ ([{"id": "spoken", "text": "Changed words", "wav": "accepted.wav"}], "changed text"),
|
|
|
+ )
|
|
|
+ for manifest, label in cases:
|
|
|
+ with self.subTest(label=label):
|
|
|
+ (self.narration / "manifest.json").unlink(missing_ok=True)
|
|
|
+ with self.assertRaises(SystemExit):
|
|
|
+ self.assemble([{"id": "spoken", "kind": "image", "src": "still.png",
|
|
|
+ "narration": "Expected words"}], manifest=manifest,
|
|
|
+ run_side_effect=lambda command: (_ for _ in ()).throw(
|
|
|
+ AssertionError("encoding started")
|
|
|
+ ))
|
|
|
+
|
|
|
+ def test_manifest_wav_name_is_authoritative_and_removed_narration_ignores_leftover_wav(self):
|
|
|
+ selected = self.narration / "accepted-name.wav"
|
|
|
+ selected.write_bytes(b"accepted")
|
|
|
+ leftover = self.narration / "silent.wav"
|
|
|
+ leftover.write_bytes(b"leftover")
|
|
|
+ (self.root / "still.png").write_bytes(b"image sentinel")
|
|
|
+ manifest = [{"id": "spoken", "text": "Expected words",
|
|
|
+ "wav": selected.name, "duration": 1.0, "synthesis": {}}]
|
|
|
+ status, calls = self.assemble([
|
|
|
+ {"id": "spoken", "kind": "image", "src": "still.png", "narration": "Expected words"},
|
|
|
+ {"id": "silent", "kind": "image", "src": "still.png"},
|
|
|
+ ], manifest=manifest)
|
|
|
+ self.assertEqual(status, 0)
|
|
|
+ encoded = [call for call in calls if call and call[0] == "ffmpeg"]
|
|
|
+ self.assertTrue(any(str(selected) in call for call in encoded))
|
|
|
+ self.assertFalse(any(str(leftover) in call for call in encoded))
|
|
|
+ offsets = json.loads((self.root / "work" / "offsets.json").read_text())
|
|
|
+ self.assertEqual(set(offsets), {"spoken"})
|
|
|
+
|
|
|
+ def test_2560x1080_movie_uses_source_audio_and_silent_source_gets_anullsrc(self):
|
|
|
+ source = self.root / "wide-2560x1080.mp4"
|
|
|
+ source.write_bytes(b"source")
|
|
|
+ status, calls = self.assemble(
|
|
|
+ [{"id": "movie", "kind": "movie", "src": source.name,
|
|
|
+ "narration": "Ignore this", "height": 800}],
|
|
|
+ run_side_effect=lambda command: subprocess.CompletedProcess(
|
|
|
+ command, 0,
|
|
|
+ json.dumps({"streams": [{"codec_type": "video", "width": 2560,
|
|
|
+ "height": 1080}]}), ""
|
|
|
+ ) if "-show_streams" in command else subprocess.CompletedProcess(command, 0, "1.0", ""),
|
|
|
+ )
|
|
|
+ self.assertEqual(status, 0)
|
|
|
+ movie_encode = next(call for call in calls if call and call[0] == "ffmpeg")
|
|
|
+ self.assertIn("anullsrc=r=44100:cl=stereo", movie_encode)
|
|
|
+ self.assertIn("0:v:0", movie_encode)
|
|
|
+ self.assertIn("1:a:0", movie_encode)
|
|
|
+ self.assertIn("scale=1920:800:force_original_aspect_ratio=decrease",
|
|
|
+ movie_encode[movie_encode.index("-vf") + 1])
|
|
|
+ offsets = json.loads((self.root / "work" / "offsets.json").read_text())
|
|
|
+ self.assertEqual(offsets, {})
|
|
|
+
|
|
|
+
|
|
|
+class PercentPathContract(unittest.TestCase):
|
|
|
+ def test_sequence_pattern_escapes_only_directory_percents(self):
|
|
|
+ module = fixtures.load_script("media_paths")
|
|
|
+ pattern = module.sequence_pattern(Path("folder%name") / "frames", "frame-%08d.png")
|
|
|
+ self.assertEqual(pattern, "folder%%name/frames/frame-%08d.png")
|
|
|
+
|
|
|
+ def test_checker_sampling_escapes_only_output_directory_percents(self):
|
|
|
+ module = fixtures.load_script("check-movie")
|
|
|
+ with tempfile.TemporaryDirectory() as directory:
|
|
|
+ work = Path(directory) / "proof%take"
|
|
|
+ work.mkdir()
|
|
|
+ command = []
|
|
|
+
|
|
|
+ def run(argv, **kwargs):
|
|
|
+ command.extend(argv)
|
|
|
+ return subprocess.CompletedProcess(argv, 0, "", "")
|
|
|
+
|
|
|
+ with patch.object(module.subprocess, "run", side_effect=run), redirect_stdout(io.StringIO()):
|
|
|
+ with self.assertRaises(SystemExit):
|
|
|
+ module.sample_picture(Path("movie.mp4"), work)
|
|
|
+ output = command[-1]
|
|
|
+ self.assertIn("proof%%take", output)
|
|
|
+ self.assertTrue(output.endswith("s%05d.png"))
|
|
|
+
|
|
|
+
|
|
|
+if __name__ == "__main__":
|
|
|
+ unittest.main()
|
Drew Ritter