test: end-to-end ingest+review integration (+ opt-in ffprobe)

Per sub-project-2 plan Task 13 / spec §10 test 8. Hermetic e2e: fake fetcher -> ingest_candidate -> proposed record; approve flips it; reload + validate_catalog; select(approved_only=True) finds it. Opt-in real-ffprobe/ffmpeg tests generate a lavfi clip and assert mode/duration/resolution + dominant color, skipped when the binaries are absent. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-04 06:35:17 -07:00
parent 376edd7819
commit deb26b2575
1 changed files with 134 additions and 0 deletions
@@ -0,0 +1,134 @@
+"""End-to-end: ingest (mocked boundaries) -> review -> select.
+
+Plus opt-in tests that invoke real ffprobe/ffmpeg, skipped when absent.
+"""
+
+import re
+import shutil
+import subprocess
+
+import pytest
+
+from hef.catalog import load_catalog, validate_catalog
+from hef.selection import Coordinate, select
+from tools.drafting import HeuristicProposer
+from tools.ingest.base import Candidate, ingest_candidate
+from tools.mediatools import compute_dominant_color
+from tools.probe import probe_file
+from tools.review import approve, proposed_records
+from tools.tagging import derive_tags
+
+
+class _FakeFetcher:
+    archive = "nasa"
+
+    def search(self, query, *, limit):
+        return [
+            Candidate(
+                source_archive="nasa",
+                source_url="https://example.org/landing",
+                media_url="https://example.org/x.mp4",
+                title="Earthrise",
+                license="public_domain",
+                attribution="",
+                suggested_id="nasa-earthrise",
+                media_ext="mp4",
+                description="earth from the moon",
+            )
+        ][:limit]
+
+    def resolve(self, identifier):
+        raise NotImplementedError
+
+
+def _av_prober(path):
+    from tools.probe import Probe
+
+    return Probe(
+        streams=[
+            {"codec_type": "video", "width": 1920, "height": 1080, "disposition": {"attached_pic": 0}},
+            {"codec_type": "audio"},
+        ],
+        format={"duration": "10.0"},
+    )
+
+
+def _fake_downloader(url, dest):
+    dest.write_bytes(b"FAKE")
+
+
+def test_end_to_end_ingest_review_select(tmp_path):
+    catalog = tmp_path / "library.jsonl"
+    media_root = tmp_path / "media"
+
+    # Ingest one candidate (all boundaries faked).
+    candidate = _FakeFetcher().search("earth", limit=1)[0]
+    rec = ingest_candidate(
+        candidate,
+        catalog_path=catalog,
+        media_root=media_root,
+        proposer=HeuristicProposer(),
+        prober=_av_prober,
+        downloader=_fake_downloader,
+    )
+    assert rec is not None and rec.review_status == "proposed"
+
+    # The proposed record can't be selected with approved_only yet.
+    loaded = load_catalog(catalog)
+    validate_catalog(loaded)
+    coord = Coordinate(loaded[0].left, loaded[0].right, loaded[0].dark, loaded[0].light)
+    assert select(loaded, coord, "av", approved_only=True) is None
+
+    # Review: approve it, persist.
+    pending = proposed_records(loaded)
+    assert len(pending) == 1
+    approved = approve(pending[0], reviewed_at="2026-06-04T13:00:00+00:00")
+    from hef.catalog import save_catalog
+
+    idx = {r.id: i for i, r in enumerate(loaded)}
+    loaded[idx[approved.id]] = approved
+    save_catalog(loaded, catalog)
+
+    # Reload, validate, and now select() finds it.
+    final = load_catalog(catalog)
+    validate_catalog(final)
+    picked = select(final, coord, "av", approved_only=True)
+    assert picked is not None and picked.id == "nasa-earthrise"
+    assert picked.review_status == "approved"
+
+
+_HAS_FFMPEG = shutil.which("ffmpeg") is not None
+_HAS_FFPROBE = shutil.which("ffprobe") is not None
+
+
+@pytest.mark.skipif(not (_HAS_FFMPEG and _HAS_FFPROBE), reason="ffmpeg/ffprobe not installed")
+def test_real_ffprobe_on_generated_clip(tmp_path):
+    clip = tmp_path / "clip.mp4"
+    subprocess.run(
+        [
+            "ffmpeg", "-v", "quiet", "-y",
+            "-f", "lavfi", "-i", "testsrc=duration=1:size=320x240:rate=30",
+            "-f", "lavfi", "-i", "sine=frequency=440:duration=1",
+            "-shortest", str(clip),
+        ],
+        check=True,
+    )
+    tags = derive_tags(probe_file(clip))
+    assert tags.mode == "av"
+    assert tags.resolution == "320x240"
+    assert tags.duration_s == 1
+
+
+@pytest.mark.skipif(not _HAS_FFMPEG, reason="ffmpeg not installed")
+def test_real_dominant_color_on_generated_clip(tmp_path):
+    clip = tmp_path / "clip.mp4"
+    subprocess.run(
+        [
+            "ffmpeg", "-v", "quiet", "-y",
+            "-f", "lavfi", "-i", "color=c=red:duration=1:size=320x240:rate=30",
+            str(clip),
+        ],
+        check=True,
+    )
+    color = compute_dominant_color(clip, midpoint_s=0.5)
+    assert re.fullmatch(r"#[0-9a-f]{6}", color)