arthur/clips/tests/test_api.py

"""What the server guarantees, as opposed to what the client intends.

The two interesting groups here are the ones that make the tier split a property
of the system rather than a convention in ClojureScript:

  A KEY DESCRIBES ITS BYTES. The server recomputes every tier-2 key it is handed
  and refuses a mismatch, so nothing can store a block under a name that is not
  the hash of its own descriptor.

  A BLOCK CAN NAME ITS DETECTOR VERSION. Every block names an analysis and every
  analysis declares a detector and a version, both enforced here. That chain is
  what docs/architecture.md asks for: without it, a model upgrade that silently
  reuses old landmarks presents as "the tool got worse" with no event to attach it
  to.

The rest is the load/save round trip, the conditional write, and the footage
manifest that makes the frames the backend's to serve.
"""
import hashlib
import json
import shutil
import struct
import subprocess
import tempfile
import zlib
from pathlib import Path
from unittest import skipUnless
from unittest.mock import Mock, patch

from django.core.files.uploadedfile import SimpleUploadedFile
from django.test import TestCase, override_settings

from clips import blobs, extraction
from clips.models import Analysis, Block, Blob, Clip, Footage, Leaf, Project, Revision, Source

BLOB_DIR = tempfile.mkdtemp(prefix="arthur-test-blobs-")


def key_for(descriptor: str) -> str:
    return "sha256:" + hashlib.sha256(descriptor.encode("utf-8")).hexdigest()


def analysis_descriptor(version="1.0.1"):
    # Canonical JSON, written the way arthur.domain.canon writes it: sorted keys,
    # no spaces, integral doubles with no point.
    return ('{"aspect":1,"detector":"mediapipe","frames":48,"fps":30,"scheme":1,'
            f'"version":"{version}"}}')


def block_descriptor(analysis_key, role="geom", anchor_avg=2):
    return (f'{{"analysis":"{analysis_key}","features":["mouth"],'
            f'"layout":{{"frames":48,"scale":16384,"stride":16,"tracks":1,"type":"int16"}},'
            f'"observation":null,"params":{{"anchor-avg":{anchor_avg}}},'
            f'"role":"{role}","scheme":1,"tracks":["outer"]}}')


def png(width=4, height=3):
    """The smallest valid PNG of a given size, written by hand.

    So that `blobs.png_size` and the ingest path are exercised without Pillow. The
    one thing the backend needs from a PNG is its IHDR, and this is a PNG with one.
    """
    def chunk(kind, payload):
        return (struct.pack(">I", len(payload)) + kind + payload
                + struct.pack(">I", zlib.crc32(kind + payload) & 0xFFFFFFFF))

    ihdr = struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0)
    raw = b"".join(b"\x00" + b"\x40\x40\x40" * width for _ in range(height))
    return (b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", ihdr)
            + chunk(b"IDAT", zlib.compress(raw)) + chunk(b"IEND", b""))


@override_settings(BLOB_ROOT=BLOB_DIR)
class BlobStoreTests(TestCase):
    def test_the_same_bytes_are_stored_once(self):
        a, size = blobs.write(b"the same bytes")
        b, _ = blobs.write(b"the same bytes")
        self.assertEqual(a, b)
        self.assertEqual(size, 14)
        self.assertEqual(blobs.read(a), b"the same bytes")

    def test_a_path_that_is_not_a_hash_is_refused(self):
        # The blob route takes its digest from the URL, so this is the check that
        # stops `/blob/../../etc/passwd` being a path at all.
        with self.assertRaises(ValueError):
            blobs.path_for("../../etc/passwd")
        with self.assertRaises(ValueError):
            blobs.path_for("deadbeef")

    def test_a_png_reports_its_own_size(self):
        with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as fh:
            fh.write(png(17, 5))
        self.assertEqual((17, 5), blobs.png_size(Path(fh.name)))

    def test_a_blob_is_served_immutable(self):
        digest, size = blobs.write(b"bytes on the wire")
        Blob.objects.create(digest=digest, size=size, media_type="application/octet-stream")
        response = self.client.get(f"/blob/{digest}")
        self.assertEqual(200, response.status_code)
        self.assertIn("immutable", response["Cache-Control"])
        self.assertEqual(f'"{digest}"', response["ETag"])
        self.assertEqual(b"bytes on the wire", b"".join(response.streaming_content))

    def test_an_unknown_blob_is_a_404_and_not_a_traceback(self):
        self.assertEqual(404, self.client.get("/blob/" + "0" * 64).status_code)
        self.assertEqual(404, self.client.get("/blob/nonsense").status_code)

    def test_a_blob_serves_byte_ranges(self):
        # NOT AN OPTIMISATION. A <video> that is handed 200 with no Accept-Ranges
        # reports an empty `seekable`, every currentTime write is a no-op, and the
        # detector then measures frame one over and over without anything raising.
        # Django's FileResponse does no Range handling, so this is the whole of
        # what makes the analysis source seekable.
        digest, _ = blobs.write(b"0123456789")
        Blob.objects.create(digest=digest, size=10, media_type="video/mp4")

        whole = self.client.get(f"/blob/{digest}")
        self.assertEqual(200, whole.status_code)
        self.assertEqual("bytes", whole["Accept-Ranges"])

        part = self.client.get(f"/blob/{digest}", headers={"range": "bytes=2-5"})
        self.assertEqual(206, part.status_code)
        self.assertEqual("bytes 2-5/10", part["Content-Range"])
        self.assertEqual("4", part["Content-Length"])
        self.assertEqual(b"2345", b"".join(part.streaming_content))

        # An open end, which is what a media element actually sends first.
        tail = self.client.get(f"/blob/{digest}", headers={"range": "bytes=7-"})
        self.assertEqual(206, tail.status_code)
        self.assertEqual("bytes 7-9/10", tail["Content-Range"])
        self.assertEqual(b"789", b"".join(tail.streaming_content))

        # A suffix range asks a different question: the LAST n bytes.
        suffix = self.client.get(f"/blob/{digest}", headers={"range": "bytes=-3"})
        self.assertEqual(206, suffix.status_code)
        self.assertEqual("bytes 7-9/10", suffix["Content-Range"])

        # Past the end is a 416 with the real length, so the client can recover.
        over = self.client.get(f"/blob/{digest}", headers={"range": "bytes=50-60"})
        self.assertEqual(416, over.status_code)
        self.assertEqual("bytes */10", over["Content-Range"])

        # Unparsable is not an error: RFC 9110 says ignore it and send it all.
        junk = self.client.get(f"/blob/{digest}", headers={"range": "furlongs=1-2"})
        self.assertEqual(200, junk.status_code)
        self.assertEqual(b"0123456789", b"".join(junk.streaming_content))


@override_settings(BLOB_ROOT=BLOB_DIR)
class Tier2Tests(TestCase):
    def post(self, url, payload):
        return self.client.post(url, data=json.dumps(payload),
                                content_type="application/json")

    def register_analysis(self, version="1.0.1"):
        descriptor = analysis_descriptor(version)
        key = key_for(descriptor)
        response = self.post("/api/analyses", {
            "key": key, "descriptor": descriptor,
            "detector": "mediapipe", "version": version,
        })
        self.assertEqual(201, response.status_code, response.content)
        return key

    def test_an_analysis_is_its_own_descriptors_hash(self):
        key = self.register_analysis()
        row = Analysis.objects.get(key=key)
        self.assertEqual("mediapipe", row.detector)
        self.assertEqual("1.0.1", row.version)
        # Idempotent: the same inputs are the same key are the same row.
        again = self.post("/api/analyses", {
            "key": key, "descriptor": analysis_descriptor(), "detector": "mediapipe",
            "version": "1.0.1",
        })
        self.assertEqual(200, again.status_code)
        self.assertEqual(1, Analysis.objects.count())

    def test_a_key_that_is_not_the_hash_of_its_descriptor_is_refused(self):
        response = self.post("/api/analyses", {
            "key": "sha256:" + "0" * 64, "descriptor": analysis_descriptor(),
        })
        self.assertEqual(409, response.status_code)
        self.assertIn("not the hash", response.json()["error"])
        self.assertEqual(0, Analysis.objects.count())

    def test_an_analysis_without_a_detector_version_is_refused(self):
        # The rule docs/architecture.md is most insistent about, enforced where a
        # client cannot forget it.
        descriptor = '{"detector":"mediapipe","frames":48,"scheme":1}'
        response = self.post("/api/analyses", {
            "key": key_for(descriptor), "descriptor": descriptor,
        })
        self.assertEqual(400, response.status_code)
        self.assertEqual("version", response.json()["missing"])

    def test_a_block_is_stored_under_the_hash_of_its_inputs(self):
        analysis = self.register_analysis()
        descriptor = block_descriptor(analysis)
        key = key_for(descriptor)
        response = self.post("/api/blocks", {
            "key": key, "descriptor": descriptor,
            "data": "AAECAwQFBgc=", "state": "AAE=",
        })
        self.assertEqual(201, response.status_code, response.content)
        row = Block.objects.get(key=key)
        self.assertEqual("geom", row.role)
        self.assertEqual(analysis, row.analysis_id)
        # Two hashes, and they are not the same hash: the key is over the inputs,
        # the blob's digest is over the bytes.
        self.assertNotEqual(key[7:], row.data.digest)
        self.assertEqual(8, row.data.size)

        fetched = self.client.get(f"/api/blocks/{key}").json()
        self.assertEqual("AAECAwQFBgc=", fetched["data"])
        self.assertEqual("AAE=", fetched["state"])
        self.assertEqual(descriptor, fetched["descriptor"])

    def test_a_block_whose_analysis_is_unknown_is_refused(self):
        descriptor = block_descriptor("sha256:" + "f" * 64)
        response = self.post("/api/blocks", {
            "key": key_for(descriptor), "descriptor": descriptor, "data": "AA==",
        })
        self.assertEqual(400, response.status_code)
        self.assertIn("analysis the server does not know", response.json()["error"])

    def test_a_block_that_does_not_say_what_its_elements_are_is_refused(self):
        analysis = self.register_analysis()
        descriptor = ('{"analysis":"%s","layout":{"frames":48},"role":"geom","scheme":1}'
                      % analysis)
        response = self.post("/api/blocks", {
            "key": key_for(descriptor), "descriptor": descriptor, "data": "AA==",
        })
        self.assertEqual(400, response.status_code)
        self.assertIn("valid readings", response.json()["error"])

    def test_only_the_missing_blocks_are_asked_for(self):
        analysis = self.register_analysis()
        here = key_for(block_descriptor(analysis))
        self.post("/api/blocks", {
            "key": here, "descriptor": block_descriptor(analysis), "data": "AA==",
        })
        elsewhere = key_for(block_descriptor(analysis, anchor_avg=3))
        response = self.post("/api/blocks/missing", {"keys": [here, elsewhere]})
        self.assertEqual([elsewhere], response.json()["missing"])

    def test_a_detector_upgrade_gives_a_block_a_new_name(self):
        # The end-to-end statement of the requirement: the same measurements under
        # a new model version are a different, additional block, and the old one is
        # unreachable from the new document rather than wrong.
        old = self.register_analysis("1.0.1")
        new_descriptor = analysis_descriptor("1.0.2")
        self.post("/api/analyses", {"key": key_for(new_descriptor),
                                    "descriptor": new_descriptor})
        for analysis in (old, key_for(new_descriptor)):
            descriptor = block_descriptor(analysis)
            self.post("/api/blocks", {"key": key_for(descriptor),
                                      "descriptor": descriptor, "data": "AAEC"})
        self.assertEqual(2, Block.objects.count())
        # One set of bytes, two names: the upgrade renamed the block and did not
        # duplicate it on disk.
        self.assertEqual(1, Blob.objects.filter(block_data_for__isnull=False).distinct().count())

    def test_an_analysis_reopens_its_three_source_blocks(self):
        analysis = self.register_analysis()
        keys = []
        for role in ("source/dense", "source/detected", "source/crops"):
            descriptor = block_descriptor(analysis, role=role)
            key = key_for(descriptor)
            self.assertEqual(201, self.post("/api/blocks", {
                "key": key, "descriptor": descriptor, "data": "AA==",
            }).status_code)
            keys.append(key)
        response = self.client.put(
            f"/api/analyses/{analysis}", json.dumps({"source_blocks": keys}),
            content_type="application/json")
        self.assertEqual(200, response.status_code, response.content)
        self.assertEqual(set(keys), set(self.client.get(
            f"/api/analyses/{analysis}").json()["source_blocks"]))
        self.assertEqual(200, self.client.put(
            f"/api/analyses/{analysis}", json.dumps({"source_blocks": keys}),
            content_type="application/json").status_code)
        self.assertEqual(400, self.client.put(
            f"/api/analyses/{analysis}", json.dumps({"source_blocks": keys[:2]}),
            content_type="application/json").status_code)


@override_settings(BLOB_ROOT=BLOB_DIR)
class DocumentTests(TestCase):
    """Tier 1: load, save, and the conditional write."""

    def setUp(self):
        self.project = Project.objects.create(name="a project")
        descriptor = analysis_descriptor()
        self.analysis = key_for(descriptor)
        self.client.post("/api/analyses", data=json.dumps(
            {"key": self.analysis, "descriptor": descriptor}),
            content_type="application/json")
        block = block_descriptor(self.analysis)
        self.block = key_for(block)
        self.client.post("/api/blocks", data=json.dumps(
            {"key": self.block, "descriptor": block, "data": "AAECAwQFBgc="}),
            content_type="application/json")

    def put(self, url, payload, **headers):
        return self.client.put(url, data=json.dumps(payload),
                               content_type="application/json", **headers)

    def leaves(self):
        # Transit-shaped, because that is what a leaf actually holds: a map with a
        # cache marker, keyword keys, and a frame-keyed inner map.
        return {
            "clip/c1/timing": ["^ ", "~:fps", 30],
            "clip/c1/timeline/main": ["^ ", "~:frames", 48],
            "clip/c1/timeline/main/node/mouth": ["^ ", "~:id", "~:mouth", "~:z", "a1"],
            "clip/c1/timeline/main/channel/mouth/geom.pts": [
                "^ ", "~:animated?", True, "~:dense",
                ["^ ", "~:store", self.block, "~:offset", 0, "~:stride", 16],
            ],
            "clip/c1/timeline/main/channel/mouth-in/vis": [
                "^ ", "~:animated?", True, "~:keys", ["^ ", "~i0", True, "~i12", False],
            ],
        }

    def save(self, leaves=None, blocks=None):
        return self.put(f"/api/projects/{self.project.id}", {
            "name": "a project",
            "clips": [{"cid": "c1", "name": "take", "analysis": self.analysis,
                       "leaves": leaves if leaves is not None else self.leaves(),
                       "blocks": blocks if blocks is not None else [self.block]}],
        })

    def test_a_document_comes_back_exactly(self):
        response = self.save()
        self.assertEqual(200, response.status_code, response.content)
        self.assertEqual(5, len(response.json()["written"]))

        loaded = self.client.get(f"/api/projects/{self.project.id}").json()
        self.assertEqual(1, len(loaded["clips"]))
        clip = loaded["clips"][0]
        self.assertEqual("c1", clip["cid"])
        self.assertEqual([self.block], clip["blocks"])
        self.assertEqual(self.analysis, clip["analysis"])
        # The whole point: byte-identical values, including the integer frame keys
        # transit writes as "~i0". A JSON round trip that stringified them would
        # come back "0" and the part would hold its first pose forever.
        self.assertEqual(self.leaves(), clip["leaves"])

    def test_an_unchanged_leaf_keeps_its_version(self):
        # What makes an entity tag worth having: a save where one channel moved
        # invalidates one leaf's etag, not the whole document's.
        self.save()
        first = {leaf.path: leaf.version for leaf in Leaf.objects.all()}
        moved = self.leaves()
        moved["clip/c1/timeline/main/channel/mouth-in/vis"] = [
            "^ ", "~:animated?", True, "~:keys", ["^ ", "~i0", False],
        ]
        response = self.save(moved)
        self.assertEqual(["clip/c1/timeline/main/channel/mouth-in/vis"], response.json()["written"])
        self.assertEqual(4, response.json()["unchanged"])
        after = {leaf.path: leaf.version for leaf in Leaf.objects.all()}
        self.assertEqual(2, after["clip/c1/timeline/main/channel/mouth-in/vis"])
        self.assertEqual(first["clip/c1/timing"], after["clip/c1/timing"])

    def test_a_removed_node_removes_its_leaf(self):
        self.save()
        fewer = {k: v for k, v in self.leaves().items()
                 if k != "clip/c1/timeline/main/node/mouth"}
        response = self.save(fewer)
        self.assertEqual(["clip/c1/timeline/main/node/mouth"], response.json()["removed"])
        self.assertEqual(4, Leaf.objects.count())

    def test_a_save_does_not_disturb_another_clip(self):
        # A save is not the only way the document changes, so a save that cleared
        # what it did not mention would undo a collaborator.
        Leaf.objects.create(project=self.project, path="clip/c2/timing", value=["^ "])
        self.save()
        self.assertTrue(Leaf.objects.filter(path="clip/c2/timing").exists())

    def test_a_leaf_addressed_to_another_clip_is_refused(self):
        response = self.save({"clip/c9/timing": ["^ "]})
        self.assertEqual(400, response.status_code)
        self.assertIn("not addressed to clip", response.json()["error"])
        self.assertEqual(0, Leaf.objects.count())

    def test_a_document_naming_blocks_the_server_lacks_is_refused(self):
        # Referential integrity across the tiers. Saved without this, the document
        # loads into a blank stage on any other machine.
        response = self.save(blocks=[self.block, "sha256:" + "a" * 64])
        self.assertEqual(409, response.status_code)
        self.assertEqual(["sha256:" + "a" * 64], response.json()["missing"])
        self.assertEqual(0, Leaf.objects.count())

    def test_every_write_bumps_the_projects_version(self):
        before = Project.objects.get(id=self.project.id).seq
        self.save()
        self.assertEqual(before + 1, Project.objects.get(id=self.project.id).seq)

    # --- the conditional write ---------------------------------------------

    def test_a_leaf_write_carries_an_etag(self):
        self.save()
        url = f"/api/projects/{self.project.id}/leaves/clip/c1/timeline/main/node/mouth"
        got = self.client.get(url)
        self.assertEqual('"1"', got["ETag"])

        ok = self.put(url, {"value": ["^ ", "~:id", "~:mouth", "~:z", "a2"]},
                      HTTP_IF_MATCH='"1"')
        self.assertEqual(200, ok.status_code)
        self.assertEqual('"2"', ok["ETag"])
        self.assertEqual(["^ ", "~:id", "~:mouth", "~:z", "a2"],
                         self.client.get(url).json()["value"])

    def test_a_stale_write_is_refused_and_says_what_is_there(self):
        # 409 with the current value, so the client can offer keep-mine /
        # take-theirs. A PUT that replaced unconditionally is the bug where the
        # loser's work disappears silently.
        self.save()
        url = f"/api/projects/{self.project.id}/leaves/clip/c1/timeline/main/node/mouth"
        self.put(url, {"value": ["^ ", "~:z", "a2"]}, HTTP_IF_MATCH='"1"')
        stale = self.put(url, {"value": ["^ ", "~:z", "a3"]}, HTTP_IF_MATCH='"1"')
        self.assertEqual(409, stale.status_code)
        self.assertEqual(2, stale.json()["version"])
        self.assertEqual(["^ ", "~:z", "a2"], stale.json()["value"])
        # And the value on the server is the one that won, not the one refused.
        self.assertEqual(["^ ", "~:z", "a2"], self.client.get(url).json()["value"])

    def test_an_unconditional_write_still_works(self):
        # Conditional writes are the protocol, not a requirement: the first write
        # of a leaf has no etag to match.
        url = f"/api/projects/{self.project.id}/leaves/clip/c1/stage"
        response = self.put(url, {"value": ["^ ", "~:width", 320]})
        self.assertEqual(200, response.status_code)
        self.assertEqual('"1"', response["ETag"])

    def test_if_match_star_requires_the_leaf_to_exist(self):
        url = f"/api/projects/{self.project.id}/leaves/clip/c1/nothing"
        self.assertEqual(409, self.put(url, {"value": []}, HTTP_IF_MATCH="*").status_code)

    # --- revisions ---------------------------------------------------------

    def test_a_revision_snapshots_the_authored_layer(self):
        self.save()
        response = self.client.post(
            f"/api/projects/{self.project.id}/revisions",
            data=json.dumps({"summary": "first pass", "author": "olive"}),
            content_type="application/json",
        )
        self.assertEqual(201, response.status_code)
        revision = Revision.objects.get()
        self.assertEqual(5, len(revision.document))
        self.assertEqual(self.leaves(), revision.document)
        # Coarse on purpose: a save does not write one, because tier 1 will hold
        # cel polygons and a snapshot per save bloats the table.
        self.save()
        self.assertEqual(1, Revision.objects.count())


@override_settings(BLOB_ROOT=BLOB_DIR)
class FootageTests(TestCase):
    """Tier 3, and the thing that makes the frames the backend's to serve: the
    manifest names every frame by URL."""

    def bundle(self, frames=3, absence=None):
        root = Path(tempfile.mkdtemp(prefix="arthur-test-bundle-"))
        (root / "frames").mkdir()
        for i in range(frames):
            (root / "frames" / f"{i + 1:04d}.png").write_bytes(png(8, 6) + bytes([i]))
        (root / "audio.wav").write_bytes(b"RIFF....WAVEfmt ")
        manifest = {"fps": 12, "frames": frames, "dir": "frames",
                    "audio": "audio.wav", "source": "IMG_8608.MOV"}
        if absence:
            manifest["feature-absence"] = absence
        (root / "manifest.json").write_text(json.dumps(manifest))
        return root

    def ingest(self, root):
        from django.core.management import call_command
        from io import StringIO

        call_command("ingest_bundle", str(root), stdout=StringIO())
        return Footage.objects.get()

    def test_a_bundle_becomes_footage_with_a_url_per_frame(self):
        footage = self.ingest(self.bundle(frames=3, absence={"eye-r": [[1, 2]]}))
        self.assertEqual(3, footage.frames)
        self.assertEqual((8, 6), (footage.width, footage.height))
        self.assertEqual(12, footage.fps)
        self.assertEqual({"eye-r": [[1, 2]]}, footage.feature_absence)

        manifest = self.client.get(f"/api/footage/{footage.id}").json()
        self.assertEqual(3, len(manifest["urls"]))
        self.assertTrue(all(url.startswith("/blob/") for url in manifest["urls"]))
        self.assertEqual(f"sha256:{footage.digest}", manifest["footage"])
        self.assertTrue(manifest["audio"].startswith("/blob/"))
        self.assertEqual({"eye-r": [[1, 2]]}, manifest["feature-absence"])
        # The frames are in order, and each one is fetchable.
        first = self.client.get(manifest["urls"][0])
        self.assertEqual(200, first.status_code)
        self.assertEqual("image/png", first["Content-Type"])

    def test_ingesting_the_same_bundle_twice_is_one_footage(self):
        root = self.bundle()
        self.ingest(root)
        self.ingest(root)
        self.assertEqual(1, Footage.objects.count())

    def test_a_bundle_whose_count_disagrees_with_its_frames_is_refused(self):
        from django.core.management import call_command
        from django.core.management.base import CommandError
        from io import StringIO

        root = self.bundle(frames=3)
        (root / "frames" / "0003.png").unlink()
        with self.assertRaisesMessage(CommandError, "refusing an inaccurate footage"):
            call_command("ingest_bundle", str(root), stdout=StringIO())

    def test_the_footage_list_does_not_carry_every_url(self):
        # A list of takes should not be a list of six hundred URLs each.
        self.ingest(self.bundle())
        listed = self.client.get("/api/footage").json()["footage"]
        self.assertEqual(1, len(listed))
        self.assertNotIn("urls", listed[0])


class PageTests(TestCase):
    def test_django_serves_the_page_at_both_urls(self):
        for url in ("/", "/index.html"):
            response = self.client.get(url)
            self.assertEqual(200, response.status_code, url)
            body = response.content.decode()
            self.assertIn("/static/arthur/js/main.js", body)
            self.assertIn("/static/mediapipe/vision_bundle.js", body)
            self.assertIn('id="app"', body)
            # The token is rendered so Django sets its cookie, which is what the
            # save path reads to write the X-CSRFToken header.
            self.assertIn("csrfmiddlewaretoken", body)

    def test_the_detector_reports_a_version_derived_from_the_model(self):
        # The version is the package version plus the model asset's own hash,
        # because a version string in the client is one somebody has to remember to
        # bump, and the server is the thing that serves the model.
        report = self.client.get("/api/detector").json()
        self.assertEqual("mediapipe", report["detector"])
        self.assertNotEqual("unknown", report["version"])
        self.assertTrue(report["model"].startswith("sha256:"))
        self.assertIn("+", report["version"])


@skipUnless(shutil.which("ffmpeg") and shutil.which("ffprobe"), "ffmpeg is required")
@override_settings(BLOB_ROOT=BLOB_DIR)
class UploadTests(TestCase):
    def test_an_ffmpeg_stage_reports_live_progress_within_its_own_span(self):
        # The job's percentage is shared between the encode and the stills, so a
        # stage reports its own fraction of its own span rather than of the job.
        # Half of the frames through a stage that owns 0-55 is 27.
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            job = Mock(progress=0)

            class FakeProcess:
                returncode = 0
                calls = 0

                def poll(self):
                    self.calls += 1
                    if self.calls == 1:
                        (root / "proxy.progress").write_text("frame=2\nprogress=continue\n")
                        return None
                    return 0

                def wait(self):
                    return 0

            with patch("clips.extraction.subprocess.Popen", return_value=FakeProcess()), \
                 patch("clips.extraction.time.sleep"):
                extraction._run_with_progress(job, ["-i", "in.mp4", "out.mp4"],
                                              root, "proxy", 4, (0, 55))
            self.assertEqual(27, job.progress)
            job.save.assert_called_once_with(update_fields=["progress", "updated"])

    def test_uploaded_video_extracts_to_reopenable_footage(self):
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / "four-frames.mp4"
            subprocess.run([
                "ffmpeg", "-hide_banner", "-loglevel", "error", "-y",
                "-f", "lavfi", "-i", "color=c=red:s=64x48:r=4:d=1",
                "-c:v", "mpeg4", str(path),
            ], check=True, capture_output=True)
            payload = path.read_bytes()

        uploaded = self.client.post("/api/sources", {
            "file": SimpleUploadedFile("four-frames.mp4", payload, content_type="video/mp4")})
        self.assertEqual(201, uploaded.status_code, uploaded.content)
        source_id = uploaded.json()["id"]
        self.assertEqual(4, uploaded.json()["probe"]["reported_frames"])
        self.assertEqual(1, Source.objects.count())
        again = self.client.post("/api/sources", {
            "file": SimpleUploadedFile("same-video.mp4", payload, content_type="video/mp4")})
        self.assertEqual(200, again.status_code, again.content)
        self.assertEqual(source_id, again.json()["id"])

        with patch("clips.extraction.enqueue", side_effect=extraction.run):
            queued = self.client.post("/api/extractions", json.dumps({
                "source": source_id, "settings": {},
            }), content_type="application/json")
        self.assertIn(queued.status_code, (200, 202), queued.content)
        job = self.client.get(f"/api/extractions/{queued.json()['key']}").json()
        self.assertEqual("done", job["state"], job)
        footage = self.client.get(f"/api/footage/{job['footage']}").json()
        self.assertEqual((4, 64, 48), (footage["frames"], footage["width"], footage["height"]))

        # THE PROXY IS THE ANALYSIS SOURCE. The page seeks this URL frame by
        # frame, so it has to exist, be a video, and answer a Range request —
        # without the last of those a media element cannot seek it at all.
        self.assertTrue(footage["video"].startswith("/blob/"), footage)
        proxy = self.client.get(footage["video"])
        self.assertEqual(200, proxy.status_code)
        self.assertEqual("video/mp4", proxy["Content-Type"])
        self.assertEqual("bytes", proxy["Accept-Ranges"])
        self.assertEqual(206, self.client.get(footage["video"],
                                              headers={"range": "bytes=0-31"}).status_code)

        # And the stills beside it are JPEGs for tracing, one per frame.
        self.assertEqual(4, len(footage["urls"]))
        still = self.client.get(footage["urls"][0])
        self.assertEqual(200, still.status_code)
        self.assertEqual("image/jpeg", still["Content-Type"])
        self.assertEqual(200, self.client.get(footage["audio"]).status_code)

    def test_the_proxy_is_re_encoded_rather_than_the_upload_re_served(self):
        # The footage's identity is the proxy's digest, and the proxy is produced
        # by one ffmpeg invocation whatever the upload was. If the upload were
        # passed through when it happened to be playable, identity would depend on
        # which branch ran — and an HEVC upload would reach a browser that cannot
        # decode it.
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / "already-h264.mp4"
            subprocess.run([
                "ffmpeg", "-hide_banner", "-loglevel", "error", "-y",
                "-f", "lavfi", "-i", "testsrc=s=64x48:r=4:d=1",
                "-c:v", "libx264", "-pix_fmt", "yuv420p", str(path),
            ], check=True, capture_output=True)
            payload = path.read_bytes()

        uploaded = self.client.post("/api/sources", {
            "file": SimpleUploadedFile("already-h264.mp4", payload, content_type="video/mp4")})
        with patch("clips.extraction.enqueue", side_effect=extraction.run):
            queued = self.client.post("/api/extractions", json.dumps({
                "source": uploaded.json()["id"], "settings": {},
            }), content_type="application/json")
        job = self.client.get(f"/api/extractions/{queued.json()['key']}").json()
        self.assertEqual("done", job["state"], job)

        footage = Footage.objects.get(id=job["footage"])
        self.assertIsNotNone(footage.video)
        self.assertNotEqual(Source.objects.get(id=uploaded.json()["id"]).blob_id,
                            footage.video_id)

    def test_a_container_whose_metadata_disagrees_with_itself_is_not_refused(self):
        # THE REGRESSION. Ordinary iPhone footage, shot straight from the camera
        # app, reports avg_frame_rate 8670/299 and nb_frames 289 on a stream whose
        # decoded timestamps are 280 frames exactly 1/30s apart. Refusing that as
        # "variable-frame-rate" rejected CFR video on the strength of a summary the
        # container got wrong about its own contents. Nothing measures the source
        # any more, so the rate is a choice rather than a fact to be verified.
        report = json.dumps({"streams": [
            {"codec_type": "video", "r_frame_rate": "30/1", "avg_frame_rate": "8670/299",
             "nb_frames": "289", "width": 1920, "height": 1440},
            {"codec_type": "audio"}],
            "format": {"duration": "9.316667"}})
        with patch("clips.extraction._command", return_value=report):
            facts = extraction.probe(Path("phone.mov"))
        self.assertEqual(30.0, facts["fps"])
        self.assertTrue(facts["vfr"], "the disagreement is still recorded, just not fatal")
        self.assertTrue(facts["has_audio"])

    def test_the_proxy_rate_is_exact_rather_than_a_rounded_float(self):
        # 30000/1001 is not a float. Handing ffmpeg's -r a rounded one is how a
        # long take drifts out of sync with its own audio.
        report = json.dumps({"streams": [
            {"codec_type": "video", "r_frame_rate": "30000/1001",
             "avg_frame_rate": "30000/1001", "width": 640, "height": 480}],
            "format": {"duration": "10"}})
        with patch("clips.extraction._command", return_value=report):
            facts = extraction.probe(Path("ntsc.mov"))
        self.assertEqual("30000/1001", facts["rate"])

    def test_a_rate_no_footage_could_have_been_shot_at_is_refused(self):
        report = json.dumps({"streams": [
            {"codec_type": "video", "r_frame_rate": "1000/1", "avg_frame_rate": "900/1",
             "width": 640, "height": 480}],
            "format": {"duration": "10"}})
        with patch("clips.extraction._command", return_value=report):
            with self.assertRaisesMessage(ValueError, "not a rate footage can be measured at"):
                extraction.probe(Path("nonsense.mov"))

    def test_variable_frame_rate_video_extracts_to_constant_rate_footage(self):
        # End to end on a genuinely variable file: irregular timestamps in, one
        # constant-rate proxy out, and the DURATION preserved — which is the thing
        # that must not move, because the page's clock is the audio.
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            held = root / "held.mp4"
            wobbly = root / "wobbly.mp4"
            subprocess.run([
                "ffmpeg", "-hide_banner", "-loglevel", "error", "-y",
                "-f", "lavfi", "-i", "testsrc=s=64x48:r=5:d=2", "-r", "30",
                "-c:v", "libx264", "-pix_fmt", "yuv420p", str(held)],
                check=True, capture_output=True)
            subprocess.run([
                "ffmpeg", "-hide_banner", "-loglevel", "error", "-y", "-i", str(held),
                "-vf", "mpdecimate", "-fps_mode", "vfr",
                "-c:v", "libx264", "-pix_fmt", "yuv420p", str(wobbly)],
                check=True, capture_output=True)
            facts = extraction.probe(wobbly)
            self.assertTrue(facts["vfr"], "the fixture is not actually variable")
            payload = wobbly.read_bytes()

        uploaded = self.client.post("/api/sources", {
            "file": SimpleUploadedFile("wobbly.mp4", payload, content_type="video/mp4")})
        self.assertEqual(201, uploaded.status_code, uploaded.content)
        with patch("clips.extraction.enqueue", side_effect=extraction.run):
            queued = self.client.post("/api/extractions", json.dumps({
                "source": uploaded.json()["id"], "settings": {},
            }), content_type="application/json")
        job = self.client.get(f"/api/extractions/{queued.json()['key']}").json()
        self.assertEqual("done", job["state"], job)

        footage = Footage.objects.get(id=job["footage"])
        self.assertEqual(facts["fps"], footage.fps)
        self.assertAlmostEqual(facts["duration"], footage.frames / footage.fps, delta=0.5)
        self.assertEqual(footage.frames, footage.frame_set.count())

    def test_footage_without_a_proxy_says_so_rather_than_serving_nothing(self):
        # Footage ingested before the proxy existed. The manifest reports a null
        # video so the loader can name the fix; it does not omit the field and let
        # the client discover it somewhere inside MediaPipe.
        audio, size = blobs.write(b"RIFF....WAVEfmt ")
        blob = Blob.objects.create(digest=audio, size=size, media_type="audio/wav")
        footage = Footage.objects.create(
            digest="e" * 64, fps=12, frames=3, width=8, height=6, audio=blob)
        manifest = self.client.get(f"/api/footage/{footage.id}").json()
        self.assertIsNone(manifest["video"])