From 3963770d5b43810bc3cd908080c1de2a620b7fee Mon Sep 17 00:00:00 2001 From: John Menke Date: Sat, 26 Sep 2026 20:56:57 -0400 Subject: [PATCH 1/2] Fail closed when an ffprobe duration probe fails or returns an unusable duration. Co-authored-by: Cursor --- src/docgen/validate.py | 74 ++++++++++++++++++++++++------ tests/test_validate_timing_sync.py | 71 ++++++++++++++++++++++++++++ 2 files changed, 132 insertions(+), 13 deletions(-) diff --git a/src/docgen/validate.py b/src/docgen/validate.py index e79031b..8f7ba00 100644 --- a/src/docgen/validate.py +++ b/src/docgen/validate.py @@ -833,9 +833,9 @@ def _check_timing_sync(self, seg_id: str) -> CheckResult: ) return CheckResult("timing_sync", True, ["Empty timing entry (skipped, non-manim)"]) - audio_dur = self._probe_media_duration(audio) - if audio_dur is None: - return CheckResult( + audio_dur, probe_fail = self._duration_or_fail(audio, "timing_sync") + if probe_fail is not None or audio_dur is None: + return probe_fail if probe_fail is not None else CheckResult( "timing_sync", False, [ @@ -947,9 +947,9 @@ def _check_story_end(self, seg_id: str) -> CheckResult: audio = self._find_audio(seg_id) if audio and not _is_lfs_pointer(audio): - audio_end = self._probe_media_duration(audio) - if audio_end is None or audio_end <= 0: - return CheckResult( + audio_end, probe_fail = self._duration_or_fail(audio, "story_end") + if probe_fail is not None or audio_end is None: + return probe_fail if probe_fail is not None else CheckResult( "story_end", False, [ @@ -1073,22 +1073,70 @@ def _timing_last_end(block: dict[str, Any]) -> float | None: return max(ends) return None + def _duration_or_fail(self, path: Path, check_name: str) -> tuple[float, None] | tuple[None, CheckResult]: + """Duration for *check_name*, or a failed check when the probe cannot be used. + + A non-zero ffprobe exit, timeout, missing binary, or unusable duration + fails the check that asked (``passed=False``). ``None`` is not a skip. + """ + try: + duration = self._probe_media_duration(path) + except RuntimeError as exc: + return None, CheckResult( + check_name, + False, + [f"cannot probe audio duration for {path.name} — {exc}"], + ) + if ( + isinstance(duration, bool) + or not isinstance(duration, (int, float)) + or not math.isfinite(float(duration)) + or float(duration) <= 0 + ): + return None, CheckResult( + check_name, + False, + [ + f"cannot probe audio duration for {path.name} — " + f"ffprobe duration is not a finite positive number: {duration!r}" + ], + ) + return float(duration), None + @staticmethod - def _probe_media_duration(path: Path) -> float | None: + def _probe_media_duration(path: Path) -> float: + """Finite positive duration from ffprobe. + + Raises ``RuntimeError`` on a non-zero exit, timeout, missing ffprobe, + or a duration that is not a finite positive number. Does not return + ``None`` for those failures (callers must fail the check). + """ try: out = subprocess.run( ["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "csv=p=0", str(path)], capture_output=True, text=True, timeout=30, ) - except (subprocess.TimeoutExpired, FileNotFoundError): - return None + except subprocess.TimeoutExpired as exc: + raise RuntimeError(f"ffprobe timed out on {path.name}") from exc + except FileNotFoundError as exc: + raise RuntimeError("ffprobe not found in PATH") from exc if out.returncode != 0: - return None + extra = (out.stderr or "").strip() + msg = f"ffprobe failed (exit {out.returncode})" + if extra: + msg = f"{msg}: {extra[:200]}" + raise RuntimeError(msg) + raw = (out.stdout or "").strip() try: - return float(out.stdout.strip()) - except ValueError: - return None + duration = float(raw) + except (TypeError, ValueError) as exc: + raise RuntimeError(f"ffprobe duration is not a number: {raw[:80]!r}") from exc + if not math.isfinite(duration) or duration <= 0: + raise RuntimeError( + f"ffprobe duration is not a finite positive number: {raw[:80]!r}" + ) + return duration # ── Helpers ──────────────────────────────────────────────────────── diff --git a/tests/test_validate_timing_sync.py b/tests/test_validate_timing_sync.py index 2775ca4..835a797 100644 --- a/tests/test_validate_timing_sync.py +++ b/tests/test_validate_timing_sync.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import subprocess import sys import types from pathlib import Path @@ -212,6 +213,76 @@ def _write_scene_spec(cfg: Config, *, labels: list[str]) -> None: (specs / "01-x.scene.yaml").write_text(yaml.dump(raw), encoding="utf-8") +class _FakeDurationProbe: + def __init__(self, returncode: int, stdout: str, stderr: str = "") -> None: + self.returncode = returncode + self.stdout = stdout + self.stderr = stderr + + +def _install_duration_probe(monkeypatch: pytest.MonkeyPatch, mode: str) -> None: + """Drive the real ``_probe_media_duration`` (do not replace the method).""" + + def _run(*_args: object, **_kwargs: object) -> _FakeDurationProbe: + if mode == "timeout": + raise subprocess.TimeoutExpired(cmd=["ffprobe"], timeout=30) + if mode == "nonzero": + # Leftover stdout looks like a duration that would pass the check. + return _FakeDurationProbe(1, "10.4\n", stderr="Invalid data") + if mode == "nan": + return _FakeDurationProbe(0, "nan\n") + if mode == "inf": + return _FakeDurationProbe(0, "inf\n") + return _FakeDurationProbe(0, "not-a-duration\n") + + monkeypatch.setattr(subprocess, "run", _run) + + +class TestMediaDurationProbeFailClosed: + """ffprobe duration failures fail the check that asked; they are not a skip.""" + + @pytest.mark.parametrize("mode", ("nonzero", "timeout", "invalid", "nan", "inf")) + def test_timing_sync_fails_closed(self, cfg: Config, monkeypatch: pytest.MonkeyPatch, mode: str) -> None: + _write_timing(cfg, last_end=10.0) + _install_duration_probe(monkeypatch, mode) + check = Validator(cfg)._check_timing_sync("01") + assert check.passed is False + assert any("cannot probe audio duration" in detail for detail in check.details) + if mode == "nonzero": + assert any("exit 1" in detail for detail in check.details) + elif mode == "timeout": + assert any("timed out" in detail for detail in check.details) + else: + assert any("ffprobe duration" in detail for detail in check.details) + + @pytest.mark.parametrize("mode", ("nonzero", "timeout", "invalid", "nan", "inf")) + def test_story_end_fails_closed(self, cfg: Config, monkeypatch: pytest.MonkeyPatch, mode: str) -> None: + words = [ + {"word": "Alpha", "start": 2.0, "end": 2.4}, + {"word": "Omega", "start": 80.0, "end": 80.5}, + ] + (cfg.animations_dir / "timing.json").write_text( + json.dumps( + { + "01-x": { + "text": "Alpha Omega", + "words": words, + "segments": [{"start": 0.0, "end": 85.0, "text": "x"}], + } + } + ), + encoding="utf-8", + ) + _write_scene_spec(cfg, labels=["Alpha", "Omega"]) + _install_duration_probe(monkeypatch, mode) + check = Validator(cfg)._check_story_end("01") + assert check.passed is False + assert any("cannot probe audio duration" in detail for detail in check.details) + if mode == "nonzero": + assert any("exit 1" in detail for detail in check.details) + assert not any("skipped" in detail.lower() for detail in check.details) + + class TestStoryEnd: def test_story_finishes_early_fails(self, cfg, monkeypatch) -> None: """Board done at ~10s while audio runs ~100s → story_end hard fail.""" From 7d5b9ef02178dd61a83d9bb1f8b4110056e6ecf5 Mon Sep 17 00:00:00 2001 From: John Menke Date: Sat, 26 Sep 2026 21:26:17 -0400 Subject: [PATCH 2/2] Keep the ffprobe failure path without raising cyclomatic complexity. --- src/docgen/validate.py | 84 ++++++++++++++++++++---------------------- 1 file changed, 40 insertions(+), 44 deletions(-) diff --git a/src/docgen/validate.py b/src/docgen/validate.py index 8f7ba00..0046bda 100644 --- a/src/docgen/validate.py +++ b/src/docgen/validate.py @@ -24,6 +24,38 @@ from docgen.config import Config +def _unusable_duration(duration: object) -> bool: + if duration is None or isinstance(duration, bool) or not isinstance(duration, (int, float)): + return True + value = float(duration) + return not math.isfinite(value) or value <= 0 + + +def _ffprobe_launch_error(path: Path, exc: BaseException) -> str: + if isinstance(exc, subprocess.TimeoutExpired): + return f"ffprobe timed out on {path.name}" + return "ffprobe not found in PATH" + + +def _positive_ffprobe_duration(out: subprocess.CompletedProcess[str]) -> float: + if out.returncode != 0: + extra = (out.stderr or "").strip() + msg = f"ffprobe failed (exit {out.returncode})" + if extra: + msg = f"{msg}: {extra[:200]}" + raise RuntimeError(msg) + raw = (out.stdout or "").strip() + try: + duration = float(raw) + except (TypeError, ValueError) as exc: + raise RuntimeError(f"ffprobe duration is not a number: {raw[:80]!r}") from exc + if not math.isfinite(duration) or duration <= 0: + raise RuntimeError( + f"ffprobe duration is not a finite positive number: {raw[:80]!r}" + ) + return duration + + @dataclass class CheckResult: name: str @@ -834,15 +866,8 @@ def _check_timing_sync(self, seg_id: str) -> CheckResult: return CheckResult("timing_sync", True, ["Empty timing entry (skipped, non-manim)"]) audio_dur, probe_fail = self._duration_or_fail(audio, "timing_sync") - if probe_fail is not None or audio_dur is None: - return probe_fail if probe_fail is not None else CheckResult( - "timing_sync", - False, - [ - f"cannot probe audio duration for {audio.name} — " - "ffprobe failed; timing_sync cannot compare the mp3 to timing.json" - ], - ) + if probe_fail is not None: + return probe_fail max_tail = float(ts_cfg.get("max_tail_gap_sec", 3.0)) max_overrun = float(ts_cfg.get("max_end_overrun_sec", 1.0)) @@ -948,15 +973,8 @@ def _check_story_end(self, seg_id: str) -> CheckResult: audio = self._find_audio(seg_id) if audio and not _is_lfs_pointer(audio): audio_end, probe_fail = self._duration_or_fail(audio, "story_end") - if probe_fail is not None or audio_end is None: - return probe_fail if probe_fail is not None else CheckResult( - "story_end", - False, - [ - f"cannot probe audio duration for {audio.name} — " - "story_end cannot compare last paced reveal to the mp3" - ], - ) + if probe_fail is not None: + return probe_fail end_t = audio_end else: # No local mp3 (or LFS pointer): compare against transcript end only. @@ -1087,12 +1105,7 @@ def _duration_or_fail(self, path: Path, check_name: str) -> tuple[float, None] | False, [f"cannot probe audio duration for {path.name} — {exc}"], ) - if ( - isinstance(duration, bool) - or not isinstance(duration, (int, float)) - or not math.isfinite(float(duration)) - or float(duration) <= 0 - ): + if _unusable_duration(duration): return None, CheckResult( check_name, False, @@ -1117,26 +1130,9 @@ def _probe_media_duration(path: Path) -> float: "-of", "csv=p=0", str(path)], capture_output=True, text=True, timeout=30, ) - except subprocess.TimeoutExpired as exc: - raise RuntimeError(f"ffprobe timed out on {path.name}") from exc - except FileNotFoundError as exc: - raise RuntimeError("ffprobe not found in PATH") from exc - if out.returncode != 0: - extra = (out.stderr or "").strip() - msg = f"ffprobe failed (exit {out.returncode})" - if extra: - msg = f"{msg}: {extra[:200]}" - raise RuntimeError(msg) - raw = (out.stdout or "").strip() - try: - duration = float(raw) - except (TypeError, ValueError) as exc: - raise RuntimeError(f"ffprobe duration is not a number: {raw[:80]!r}") from exc - if not math.isfinite(duration) or duration <= 0: - raise RuntimeError( - f"ffprobe duration is not a finite positive number: {raw[:80]!r}" - ) - return duration + except (subprocess.TimeoutExpired, FileNotFoundError) as exc: + raise RuntimeError(_ffprobe_launch_error(path, exc)) from exc + return _positive_ffprobe_duration(out) # ── Helpers ────────────────────────────────────────────────────────