superpowers/tests/proving-it-works-with-a-movie/test_narration_contract.py

import base64
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 ChatResponseContract(unittest.TestCase):
    def test_malformed_transcripts_cannot_publish_candidates_in_any_asr_mode(self):
        cases = ({}, {"transcript": None}, {"transcript": 42},
                 {"transcript": ["Two", "words"]}, {"transcript": {}},
                 {"transcript": False}, {"transcript": ""},
                 {"transcript": "   "}, {"transcript": "...!?"})
        for response in cases:
            for mode in ("off", "auto", "on"):
                with self.subTest(response=response, mode=mode), tempfile.TemporaryDirectory() as temp:
                    module = fixtures.load_script("narrate")
                    root = Path(temp)
                    scenes, output = root / "scenes.json", root / "narration"
                    scenes.write_text(json.dumps({"scenes": [{"id": "clip", "narration": "Two words"}]}))
                    sentinels = []

                    def post(*args, **kwargs):
                        sentinel = f"NOT MEDIA: response {len(sentinels) + 1}".encode()
                        sentinels.append(sentinel)
                        audio = dict(response, data=base64.b64encode(sentinel).decode())
                        return {"choices": [{"message": {"audio": audio}}]}

                    argv = ["narrate", str(scenes), str(output), "--engine", "openai-chat", "--verify", mode]
                    with patch.object(sys, "argv", argv), \
                         patch.object(module.shutil, "which", return_value="fake-ffprobe"), \
                         patch.object(module, "openai_key", return_value="fake-key"), \
                         patch.object(module, "post", post), \
                         patch.object(module, "duration", return_value=1.25), \
                         patch.object(module, "transcribe_local", side_effect=AssertionError("invalid transcript reached ASR")), \
                         redirect_stdout(io.StringIO()), redirect_stderr(io.StringIO()):
                        try:
                            result = module.main()
                        except Exception as error:
                            result = error
                    self.assertEqual(result, 1)
                    self.assertEqual(json.loads((output / "manifest.json").read_text()), [])
                    self.assertEqual(len(sentinels), 2)
                    self.assertEqual({path.read_bytes() for path in output.glob(".clip.attempt-*.wav")}, set(sentinels))
                    self.assertFalse((output / "clip.wav").exists())

    def test_valid_chat_and_cached_reuse_obey_independent_asr_modes(self):
        for mode in ("off", "auto", "on"):
            for heard in (None, "Two words"):
                with self.subTest(mode=mode, heard=heard), tempfile.TemporaryDirectory() as temp:
                    module = fixtures.load_script("narrate")
                    root = Path(temp)
                    scenes, output = root / "scenes.json", root / "narration"
                    scenes.write_text(json.dumps({"scenes": [{"id": "clip", "narration": "Two words"}]}))
                    sentinel = b"NOT MEDIA: accepted response"
                    response = {"choices": [{"message": {"audio": {
                        "data": base64.b64encode(sentinel).decode(), "transcript": "Two words",
                    }}}]}
                    argv = ["narrate", str(scenes), str(output), "--engine", "openai-chat", "--verify", "off"]
                    with patch.object(sys, "argv", argv), \
                         patch.object(module.shutil, "which", return_value="fake-ffprobe"), \
                         patch.object(module, "openai_key", return_value="fake-key"), \
                         patch.object(module, "post", return_value=response) as post, \
                         patch.object(module, "duration", return_value=1.25), \
                         patch.object(module, "transcribe_local", return_value=heard) as asr, \
                         redirect_stdout(io.StringIO()), redirect_stderr(io.StringIO()):
                        self.assertEqual(module.main(), 0)
                        argv[-1] = mode
                        expected = 1 if mode == "on" and heard is None else 0
                        self.assertEqual(module.main(), expected)
                    self.assertEqual(post.call_count, 1, "accepted cache must not synthesize again")
                    self.assertEqual(asr.call_count, int(mode != "off"))
                    self.assertEqual(bool(json.loads((output / "manifest.json").read_text())), expected == 0)
                    self.assertEqual((output / "clip.wav").read_bytes(), sentinel)


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, engine="piper",
                    extra_options=(), transcript=None, asr_calls=None):
        self.scenes.write_text(json.dumps({"scenes": scenes}), encoding="utf-8")
        argv = ["narrate", str(self.scenes), str(self.output), "--engine", engine,
                "--verify", verify, *extra_options]
        stderr = io.StringIO()
        def transcribe(wav, model):
            if asr_calls is not None:
                asr_calls.append((wav, model))
            return transcript
        with patch.object(sys, "argv", argv), \
             patch.object(self.module.shutil, "which", return_value="ffprobe"), \
             patch.object(self.module, "openai_key", return_value="key" if engine.startswith("openai") else None), \
             patch.object(self.module, "say_piper", side_effect=synthesize), \
             patch.object(self.module, "say_openai_chat",
                          side_effect=lambda key, text, wav, voice: synthesize(text, wav, voice)), \
             patch.object(self.module, "duration", return_value=1.25), \
             patch.object(self.module, "transcribe_local", side_effect=transcribe), \
             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"}]
        def accepted(text, wav, voice):
            wav.write_bytes(b"accepted")
            return text

        self.run_narrate(scenes, accepted, engine="openai-chat")
        first = self.run_narrate(scenes, synthesize, engine="openai-chat", expected=1, extra_options=("--force",))
        attempts_after_rejection = len(attempts)
        second = self.run_narrate(scenes, synthesize, engine="openai-chat", expected=1)

        self.assertEqual(first, [])
        self.assertEqual(second, [])
        self.assertGreater(len(attempts), attempts_after_rejection)
        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))
        self.assertEqual(len({path.read_bytes() for path in rejected_paths}), len(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_duration_failures_keep_distinct_attempt_bytes_across_reruns(self):
        scenes = [{"id": "clip", "narration": "Measure this clip"}]

        def synthesize(text, wav, voice):
            wav.write_bytes(f"attempt {len(list(self.output.glob('.clip.attempt-*.wav'))) + 1}".encode())

        for expected_attempts in (1, 2):
            self.scenes.write_text(json.dumps({"scenes": scenes}), encoding="utf-8")
            argv = ["narrate", str(self.scenes), str(self.output), "--engine", "piper", "--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=None), \
                 patch.object(self.module, "say_piper", side_effect=synthesize), \
                 patch.object(self.module, "duration", side_effect=RuntimeError("ffprobe failed")), \
                 redirect_stdout(io.StringIO()), redirect_stderr(io.StringIO()):
                self.assertEqual(self.module.main(), 1)
            attempts = sorted(self.output.glob(".clip.attempt-*.wav"))
            self.assertEqual(len(attempts), expected_attempts)
            self.assertEqual(len({path.read_bytes() for path in attempts}), expected_attempts)
            self.assertFalse((self.output / "clip.wav").exists())

    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):
                asr_calls = []
                manifest = self.run_narrate(scenes, synthesize, verify=mode,
                                             expected=expected, transcript="\u77ed\u6587",
                                             asr_calls=asr_calls)
                self.assertEqual(bool(manifest), expected == 0)
                self.assertEqual(bool(asr_calls), mode != "off")

    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_fresh_unsupported_chat_transcript_is_rejected_before_synthesis(self):
        called = []

        def synthesize(*args):
            called.append(args)

        manifest = self.run_narrate(
            [{"id": "clip", "narration": "\U00020000\U00020001"}], synthesize,
            engine="openai-chat", expected=1,
        )
        self.assertEqual(manifest, [])
        self.assertEqual(called, [])

    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")
        for mode in ("off", "auto"):
            with self.subTest(mode=mode):
                (self.output / "manifest.json").write_text(json.dumps([{
                    "id": "clip", "text": "\u77ed\u6587", "wav": "clip.wav", "duration": 1.0,
                    "synthesis": synthesis,
                }]), encoding="utf-8")
                argv = ["narrate", str(self.scenes), str(self.output), "--engine", "openai-chat",
                        "--verify", mode]
                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()), [])

    def test_empty_chat_or_asr_speech_is_rejected(self):
        def empty_chat_synthesis(text, wav, voice):
            wav.write_bytes(b"audio")
            return ""

        empty_chat = self.run_narrate(
            [{"id": "clip", "narration": "Two words"}],
            empty_chat_synthesis, engine="openai-chat", expected=1,
        )
        self.assertEqual(empty_chat, [])

        def synthesize(text, wav, voice):
            wav.write_bytes(b"audio")

        self.scenes.write_text(json.dumps({"scenes": [{"id": "clip", "narration": "Two words"}]}),
                              encoding="utf-8")
        argv = ["narrate", str(self.scenes), str(self.output), "--engine", "piper", "--verify", "auto"]
        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, "transcribe_local", return_value=""), \
             patch.object(self.module, "duration", return_value=1.0), \
             redirect_stdout(io.StringIO()), redirect_stderr(io.StringIO()):
            self.assertEqual(self.module.main(), 1)
        self.assertEqual(json.loads((self.output / "manifest.json").read_text()), [])

    def test_cached_nested_wav_keeps_its_manifest_path_after_reverification(self):
        nested = self.output / "takes" / "clip.wav"
        nested.parent.mkdir(parents=True)
        nested.write_bytes(b"accepted")
        synthesis = {"engine": "piper", "voice": self.module.PIPER_VOICE,
                     "model": self.module.PIPER_VOICE}
        (self.output / "manifest.json").write_text(json.dumps([{
            "id": "clip", "text": "Nested clip", "wav": "takes/clip.wav", "duration": 1.0,
            "synthesis": synthesis,
        }]), encoding="utf-8")
        manifest = self.run_narrate(
            [{"id": "clip", "narration": "Nested clip"}],
            lambda *args: (_ for _ in ()).throw(AssertionError("cached")), verify="on",
            transcript="Nested clip",
        )
        self.assertEqual(manifest[0]["wav"], "takes/clip.wav")

    def test_cached_strict_verification_withdraws_before_interrupt(self):
        def synthesize(text, wav, voice):
            wav.write_bytes(b"accepted")

        self.run_narrate([{"id": "clip", "narration": "Interrupt safely"}], synthesize)
        self.scenes.write_text(json.dumps({"scenes": [{"id": "clip", "narration": "Interrupt safely"}]}),
                              encoding="utf-8")
        argv = ["narrate", str(self.scenes), str(self.output), "--engine", "piper", "--verify", "on"]
        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=AssertionError("cached")), \
             patch.object(self.module, "transcribe_local", side_effect=KeyboardInterrupt), \
             redirect_stdout(io.StringIO()), redirect_stderr(io.StringIO()):
            with self.assertRaises(KeyboardInterrupt):
                self.module.main()
        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"),
            ("\U00020000\U00020001", "\U00020002\U00020003"),
            ("*** !!!", "*** !!!"),
        ):
            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, {})

    def test_movie_with_audio_maps_its_source_audio(self):
        source = self.root / "source-with-audio.mp4"
        source.write_bytes(b"source")
        status, calls = self.assemble(
            [{"id": "movie", "kind": "movie", "src": source.name}],
            run_side_effect=lambda command: subprocess.CompletedProcess(
                command, 0,
                json.dumps({"streams": [{"codec_type": "video"}, {"codec_type": "audio"}]}), ""
            ) 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.assertNotIn("anullsrc=r=44100:cl=stereo", movie_encode)
        self.assertEqual(movie_encode[movie_encode.index("-map") + 1], "0:v:0")
        self.assertEqual(movie_encode[movie_encode.index("-map", movie_encode.index("-map") + 1) + 1], "0:a:0")

    def test_movie_geometry_fits_width_and_requested_inner_height_before_padding(self):
        self.assertEqual(self.module.movie_geometry(1920, 1080, 800), {
            "scale": (1920, 800),
            "pad": (1920, 1080),
        })


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()