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 base64
import hashlib
import json
import shutil
import struct
import subprocess
import tempfile
import zlib
from io import StringIO
from pathlib import Path
from unittest import skipUnless
from unittest.mock import Mock, patch

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

from clips import blobs, extraction
from clips.models import Analysis, Block, Blob, Clip, Footage, Image, Leaf, Project, Revision, Sound, 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"])

    @override_settings(DATA_UPLOAD_MAX_MEMORY_SIZE=1024, FILE_UPLOAD_MAX_MEMORY_SIZE=1024)
    def test_large_block_upload_streams_past_json_body_limit(self):
        analysis = self.register_analysis()
        descriptor = block_descriptor(analysis, role="source/crops")
        key = key_for(descriptor)
        payload = bytes(range(256)) * 16
        response = self.client.post("/api/blocks", {
            "key": key,
            "descriptor": descriptor,
            "data": SimpleUploadedFile("block.bin", payload),
            "state": SimpleUploadedFile("state.bin", b"\x00\x01"),
        })
        self.assertEqual(201, response.status_code, response.content)
        row = Block.objects.get(key=key)
        self.assertEqual(blobs.CROP_MEDIA_TYPE, row.data.media_type)
        self.assertLess(row.data.size, len(payload))
        self.assertEqual(payload, zlib.decompress(blobs.read(row.data_id)))
        self.assertEqual(base64.b64encode(payload).decode(),
                         self.client.get(f"/api/blocks/{key}").json()["data"])
        self.assertEqual(b"\x00\x01", blobs.read(row.state_id))

    def test_existing_raw_crop_block_is_compressed_without_changing_its_key_or_read(self):
        analysis = self.register_analysis()
        descriptor = block_descriptor(analysis, role="source/crops")
        key = key_for(descriptor)
        payload = b"raw crop pixels" * 100
        digest, size = blobs.write(payload)
        old = Blob.objects.create(digest=digest, size=size)
        Block.objects.create(key=key, descriptor=descriptor, role="source/crops",
                             analysis_id=analysis, data=old)

        call_command("compress_crop_blocks", stdout=StringIO())
        row = Block.objects.select_related("data").get(key=key)
        self.assertEqual(blobs.CROP_MEDIA_TYPE, row.data.media_type)
        self.assertEqual(base64.b64encode(payload).decode(),
                         self.client.get(f"/api/blocks/{key}").json()["data"])
        self.assertFalse(blobs.path_for(digest).exists())
        compressed_digest = row.data_id
        call_command("compress_crop_blocks", stdout=StringIO())
        self.assertEqual(compressed_digest, Block.objects.get(key=key).data_id)

    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)

    def test_source_roles_are_complete_and_unique_per_subject(self):
        analysis = self.register_analysis()
        keys = []
        for subject in ("face-1", "face-2"):
            for role in ("source/dense", "source/detected", "source/crops"):
                desc = json.loads(block_descriptor(analysis, role=role))
                desc["features"] = [subject]
                descriptor = json.dumps(desc, sort_keys=True, separators=(",", ":"))
                key = key_for(descriptor)
                self.assertEqual(201, self.post("/api/blocks", {
                    "key": key, "descriptor": descriptor, "data": "AA==",
                }).status_code)
                keys.append(key)

        def put(keys):
            return self.client.put(f"/api/analyses/{analysis}",
                                   json.dumps({"source_blocks": keys}),
                                   content_type="application/json")

        self.assertEqual(400, put(keys[:-1]).status_code)
        self.assertEqual(400, put(keys + keys[:1]).status_code)
        self.assertEqual(200, put(keys).status_code)
        self.assertEqual(200, put(list(reversed(keys))).status_code)
        self.assertEqual(409, put(keys[:3]).status_code)
        self.assertEqual(set(keys), set(self.client.get(
            f"/api/analyses/{analysis}").json()["source_blocks"]))



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

    def setUp(self):
        from django.contrib.auth import get_user_model
        owner = get_user_model().objects.create_user("owner", password="password1")
        self.client.force_login(owner)
        self.project = Project.objects.create(name="a project", owner=owner)
        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/symbol/main": ["^ ", "~:frames", 48],
            "clip/c1/symbol/main/node/mouth": ["^ ", "~:id", "~:mouth", "~:z", "a1"],
            "clip/c1/symbol/main/channel/mouth/geom.pts": [
                "^ ", "~:animated?", True, "~:dense",
                ["^ ", "~:store", self.block, "~:offset", 0, "~:stride", 16],
            ],
            "clip/c1/symbol/main/channel/mouth-in/vis": [
                "^ ", "~:animated?", True, "~:keys", ["^ ", "~i0", True, "~i12", False],
            ],
        }

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

    def test_a_clip_declares_the_registered_analyses_its_blocks_name(self):
        undeclared = self.save(analyses=[])
        self.assertEqual(409, undeclared.status_code)
        self.assertIn("every block", undeclared.json()["error"])

        unknown = "sha256:" + "f" * 64
        missing = self.save(analyses=[self.analysis, unknown])
        self.assertEqual(409, missing.status_code)
        self.assertEqual([unknown], missing.json()["missing"])

    def test_every_saved_symbol_is_listed_across_projects(self):
        leaves = self.leaves()
        leaves["clip/c1/symbol/sym~face"] = ["^ ", "~:name", "face", "~:frames", 12]
        leaves["clip/c1/symbol/sym~face/node/mark"] = ["^ ", "~:id", "~:mark", "~:z", "a1"]
        self.assertEqual(200, self.save(leaves).status_code)
        rows = self.client.get("/api/symbols").json()["symbols"]
        self.assertEqual(
            [("face", "sym~face", 12), ("main", "main", 48)],
            [(r["name"], r["symbol"], r["frames"]) for r in rows])
        self.assertEqual({str(self.project.id)}, {r["project"] for r in rows})
        self.assertEqual({"c1"}, {r["cid"] for r in rows})

    def test_saved_palettes_are_listed_as_assets(self):
        leaves = self.leaves()
        leaves["clip/c1/palette/night"] = [
            "^ ", "~:id", "~:night", "~:name", "Moonlit",
            "~:slots", ["~#list", [["^ ", "~:hex", "#001122"]]],
        ]
        self.assertEqual(200, self.save(leaves).status_code)
        rows = self.client.get("/api/symbols").json()["palettes"]
        self.assertEqual(
            [("Moonlit", "night", "c1")],
            [(r["name"], r["palette"], r["cid"]) for r in rows],
        )

    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(7, loaded["schema_version"])
        self.assertEqual(1, len(loaded["clips"]))
        clip = loaded["clips"][0]
        self.assertEqual("c1", clip["cid"])
        self.assertEqual([self.block], clip["blocks"])
        self.assertNotIn("analysis", clip)
        # 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/symbol/main/channel/mouth-in/vis"] = [
            "^ ", "~:animated?", True, "~:keys", ["^ ", "~i0", False],
        ]
        response = self.save(moved)
        self.assertEqual(["clip/c1/symbol/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/symbol/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/symbol/main/node/mouth"}
        response = self.save(fewer)
        self.assertEqual(["clip/c1/symbol/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/symbol/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/symbol/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 patch(self, base, leaves, removed=()):
        return self.put(f"/api/projects/{self.project.id}", {
            "base": base,
            "clips": [{"cid": "c1", "analyses": [self.analysis], "leaves": leaves,
                       "removed": list(removed), "blocks": [self.block]}],
        })

    def test_a_patch_leaves_what_it_does_not_name_alone(self):
        seq = self.save().json()["seq"]
        response = self.patch(seq, {"clip/c1/timing": ["^ ", "~:fps", 24]},
                              removed=["clip/c1/symbol/main/node/mouth"])
        self.assertEqual(200, response.status_code, response.content)
        self.assertEqual(["clip/c1/timing"], response.json()["written"])
        self.assertEqual(4, Leaf.objects.count())

    def test_two_people_on_different_leaves_both_land(self):
        seq = self.save().json()["seq"]
        self.assertEqual(200, self.patch(seq, {"clip/c1/timing": ["^ ", "~:fps", 24]}).status_code)
        # The second saver has not caught up, and touched a different leaf.
        response = self.patch(seq, {"clip/c1/symbol/main": ["^ ", "~:frames", 12]})
        self.assertEqual(200, response.status_code, response.content)
        leaves = self.client.get(f"/api/projects/{self.project.id}").json()["clips"][0]["leaves"]
        self.assertEqual(["^ ", "~:fps", 24], leaves["clip/c1/timing"])
        self.assertEqual(["^ ", "~:frames", 12], leaves["clip/c1/symbol/main"])

    def test_two_people_on_one_leaf_is_a_conflict_that_writes_nothing(self):
        seq = self.save().json()["seq"]
        self.patch(seq, {"clip/c1/timing": ["^ ", "~:fps", 24]})
        response = self.patch(seq, {"clip/c1/timing": ["^ ", "~:fps", 12],
                                    "clip/c1/symbol/main": ["^ ", "~:frames", 12]})
        self.assertEqual(409, response.status_code)
        self.assertEqual({"clip/c1/timing": ["^ ", "~:fps", 24]}, response.json()["conflicts"])
        self.assertEqual(seq + 1, Project.objects.get(id=self.project.id).seq)
        self.assertEqual(["^ ", "~:frames", 48],
                         Leaf.objects.get(path="clip/c1/symbol/main").value)
        # Caught up to their seq, the same write is ordinary.
        self.assertEqual(200, self.patch(seq + 1, {"clip/c1/timing": ["^ ", "~:fps", 12]}).status_code)

    def test_a_named_snapshot_restores_as_an_ordinary_write(self):
        self.save()
        snap = self.client.post(f"/api/projects/{self.project.id}/revisions",
                                data=json.dumps({"summary": "before the big change"}),
                                content_type="application/json").json()
        moved = self.leaves()
        moved["clip/c1/timing"] = ["^ ", "~:fps", 12]
        del moved["clip/c1/symbol/main/node/mouth"]
        self.save(moved)
        listed = self.client.get(f"/api/projects/{self.project.id}/revisions").json()["revisions"]
        self.assertEqual(["before the big change"], [r["summary"] for r in listed])
        restored = self.client.post(
            f"/api/projects/{self.project.id}/revisions/{snap['id']}/restore").json()
        self.assertEqual(2, restored["changed"])
        leaves = self.client.get(f"/api/projects/{self.project.id}").json()["clips"][0]["leaves"]
        self.assertEqual(self.leaves(), leaves)

    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_footage_can_be_renamed_and_falls_back_when_cleared(self):
        """A LABEL IS THE ONE FIELD A CLIENT MAY WRITE ON FOOTAGE. The rest is a
        description of bytes that are content-addressed and immutable, so a
        rename that could reach `frames` or `digest` would let the pool's name
        for a clip contradict the clip."""
        footage = self.ingest(self.bundle())
        self.assertEqual("IMG_8608.MOV", self.client.get(
            f"/api/footage/{footage.id}").json()["label"])

        renamed = self.client.patch(f"/api/footage/{footage.id}",
                                    json.dumps({"label": "  the long take  "}),
                                    content_type="application/json")
        self.assertEqual(200, renamed.status_code, renamed.content)
        self.assertEqual("the long take", renamed.json()["label"])
        # On the row, so every project listing this footage sees the new name.
        footage.refresh_from_db()
        self.assertEqual("the long take", footage.label)
        self.assertEqual("the long take",
                         self.client.get("/api/footage").json()["footage"][0]["label"])

        # Cleared gives back the name it was ingested under rather than nothing.
        cleared = self.client.patch(f"/api/footage/{footage.id}",
                                    json.dumps({"label": ""}),
                                    content_type="application/json")
        self.assertEqual("IMG_8608.MOV", cleared.json()["label"])
        self.assertEqual(3, Footage.objects.get().frames)

    def test_a_rename_that_names_no_label_is_refused(self):
        footage = self.ingest(self.bundle())
        refused = self.client.patch(f"/api/footage/{footage.id}",
                                    json.dumps({"frames": 900}),
                                    content_type="application/json")
        self.assertEqual(400, refused.status_code)
        self.assertEqual("a rename needs a label", refused.json()["error"])
        self.assertEqual(3, Footage.objects.get().frames)

    def test_a_sound_is_renamed_without_losing_the_name_it_arrived_as(self):
        """`filename` is a fact about the upload and `label` is what a person
        called it, which is why renaming does not write over the first one."""
        digest, size = blobs.write_stream([b"RIFF....WAVEfmt "])
        blob = Blob.objects.create(digest=digest, size=size, media_type="audio/wav")
        sound = Sound.objects.create(blob=blob, filename="rec0012.wav", duration=2.5)

        renamed = self.client.patch(f"/api/sounds/{sound.id}",
                                    json.dumps({"label": "arthur, line 4"}),
                                    content_type="application/json")
        self.assertEqual(200, renamed.status_code, renamed.content)
        self.assertEqual("arthur, line 4", renamed.json()["label"])
        self.assertEqual("rec0012.wav", renamed.json()["filename"])
        sound.refresh_from_db()
        self.assertEqual("rec0012.wav", sound.filename)
        self.assertEqual("arthur, line 4", sound.label)

        self.client.patch(f"/api/sounds/{sound.id}", json.dumps({"label": "   "}),
                          content_type="application/json")
        self.assertEqual("rec0012.wav",
                         self.client.get(f"/api/sounds/{sound.id}").json()["label"])

    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_an_uploaded_mp3_is_a_sound_and_not_footage(self):
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / "tone.mp3"
            subprocess.run([
                "ffmpeg", "-hide_banner", "-loglevel", "error", "-y",
                "-f", "lavfi", "-i", "sine=frequency=440:duration=1.5", str(path),
            ], check=True, capture_output=True)
            payload = path.read_bytes()

        uploaded = self.client.post("/api/sounds", {
            "file": SimpleUploadedFile("tone.mp3", payload, content_type="audio/mpeg")})
        self.assertEqual(201, uploaded.status_code, uploaded.content)
        sound = uploaded.json()
        self.assertEqual("tone.mp3", sound["label"])
        self.assertAlmostEqual(1.5, sound["duration"], delta=0.1)
        self.assertEqual(payload, b"".join(self.client.get(sound["audio"]).streaming_content))
        self.assertEqual(sound, self.client.get(f"/api/sounds/{sound['id']}").json())
        self.assertEqual([sound], self.client.get("/api/sounds").json()["sounds"])
        self.assertEqual(0, Source.objects.count())

        again = self.client.post("/api/sounds", {
            "file": SimpleUploadedFile("again.mp3", payload, content_type="audio/mpeg")})
        self.assertEqual(200, again.status_code)
        self.assertEqual(1, Sound.objects.count())

    def test_an_uploaded_still_is_an_image_named_by_its_bytes(self):
        payload = png(17, 5)
        uploaded = self.client.post("/api/images", {
            "file": SimpleUploadedFile("sheet.png", payload, content_type="image/png")})
        self.assertEqual(201, uploaded.status_code, uploaded.content)
        image = uploaded.json()
        self.assertEqual(("sheet.png", 17, 5), (image["label"], image["width"], image["height"]))
        self.assertEqual(f"/blob/{image['digest']}", image["url"])
        self.assertEqual(payload, b"".join(self.client.get(image["url"]).streaming_content))
        self.assertEqual([image], self.client.get("/api/images").json()["images"])

        again = self.client.post("/api/images", {
            "file": SimpleUploadedFile("again.png", payload, content_type="image/png")})
        self.assertEqual(200, again.status_code)
        self.assertEqual(1, Image.objects.count())

        renamed = self.client.patch(f"/api/images/{image['id']}",
                                    json.dumps({"label": "model sheet"}),
                                    content_type="application/json")
        self.assertEqual("model sheet", renamed.json()["label"])

    def test_a_file_that_is_not_a_picture_is_not_an_image(self):
        refused = self.client.post("/api/images", {
            "file": SimpleUploadedFile("notes.txt", b"not a picture", content_type="text/plain")})
        self.assertEqual(400, refused.status_code)

    def test_a_file_without_audio_is_not_a_sound(self):
        refused = self.client.post("/api/sounds", {
            "file": SimpleUploadedFile("notes.txt", b"not audio", content_type="text/plain")})
        self.assertEqual(400, refused.status_code)

    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_re_uploading_a_source_re_reads_its_facts(self):
        # A SOURCE ROW HOLDS A READING, NOT A DECISION. The facts are a pure
        # function of bytes that are themselves this row's identity, so the row
        # cannot be the place a reading goes to be preserved: `probe` got better
        # at phone footage — it stopped believing a declared 120 over timestamps
        # 1/30s apart — and a stored reading that nothing can replace would have
        # left every already-uploaded source resampling to four times the frames
        # with no way to correct it short of deleting the row.
        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()

        first = self.client.post("/api/sources", {
            "file": SimpleUploadedFile("four-frames.mp4", payload, content_type="video/mp4")})
        self.assertEqual(201, first.status_code, first.content)
        self.assertEqual(4.0, first.json()["probe"]["fps"])

        better = dict(first.json()["probe"], fps=12.0, rate="12/1", measured_fps=12.0)
        with patch("clips.extraction.probe", return_value=better):
            again = self.client.post("/api/sources", {
                "file": SimpleUploadedFile("same.mp4", payload, content_type="video/mp4")})
        self.assertEqual(200, again.status_code, again.content)
        self.assertFalse(again.json()["created"], "the same bytes are the same source")
        self.assertEqual("12/1", again.json()["probe"]["rate"])
        self.assertEqual(12.0, Source.objects.get(id=first.json()["id"]).probe["fps"])

    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_a_declared_rate_the_timestamps_do_not_bear_out_is_not_resampled_to(self):
        # THE FOUR-TIMES. An iPhone container declares `r_frame_rate` 120 over a
        # stream whose frames are 1/30s apart, and taking the declaration at its
        # word turned an 11-second clip into 1293 proxy frames instead of 323:
        # four times the encode, four times the tracing stills, four times the
        # blobs and the rows, for 970 frames that are copies of their neighbours.
        # The timestamps are the evidence and they say 30.
        streams = json.dumps({"streams": [
            {"codec_type": "video", "r_frame_rate": "120/1",
             "avg_frame_rate": "96900/3233", "nb_frames": "323",
             "width": 1920, "height": 1440},
            {"codec_type": "audio"}],
            "format": {"duration": "10.775"}})
        # IN DECODE ORDER, which is how an HEVC stream really arrives — the first
        # packets of the fixture this was found on come out 0, 0.133, 0.067,
        # 0.033. Differencing that order unsorted measures the reordering delay
        # and not the rate, so the fixture keeps the hazard in it.
        shuffled = [0, 4, 2, 1, 3, 8, 6, 5, 7, 12, 10, 9, 11]
        packets = json.dumps({"packets": [{"pts_time": f"{i / 30:.6f}"} for i in shuffled]})
        with patch("clips.extraction._command", side_effect=[streams, packets]):
            facts = extraction.probe(Path("phone.mov"))
        self.assertEqual("96900/3233", facts["rate"], "resampled to the declared 120")
        self.assertAlmostEqual(30.0, facts["measured_fps"], places=2)

    def test_a_genuine_high_rate_capture_is_still_taken_at_its_own_rate(self):
        # The other half of the same decision, and the one that would be easy to
        # break: a real 120fps capture must not be dragged down to anything. Its
        # declaration and its timestamps agree, so the declaration — the exact
        # rational the stream was authored at — is what is used.
        streams = json.dumps({"streams": [
            {"codec_type": "video", "r_frame_rate": "120/1", "avg_frame_rate": "120/1",
             "width": 640, "height": 480}],
            "format": {"duration": "2"}})
        packets = json.dumps({"packets": [{"pts_time": f"{i / 120:.6f}"} for i in range(13)]})
        with patch("clips.extraction._command", side_effect=[streams, packets]):
            facts = extraction.probe(Path("slowmo.mov"))
        self.assertEqual("120/1", facts["rate"])

    def test_too_few_timestamps_to_measure_leaves_the_declaration_standing(self):
        # A clip with nine-ish frames cannot outvote one odd timestamp, so the
        # measurement declines to have an opinion and the nominal rate — the one
        # that drops no distinct frame — is used exactly as it was before.
        streams = json.dumps({"streams": [
            {"codec_type": "video", "r_frame_rate": "30/1", "avg_frame_rate": "24/1",
             "width": 640, "height": 480}],
            "format": {"duration": "0.1"}})
        packets = json.dumps({"packets": [{"pts_time": f"{i / 30:.6f}"} for i in range(3)]})
        with patch("clips.extraction._command", side_effect=[streams, packets]):
            facts = extraction.probe(Path("tiny.mov"))
        self.assertEqual("30/1", facts["rate"])
        self.assertIsNone(facts["measured_fps"])

    def test_a_rate_measurement_that_fails_outright_cannot_refuse_an_upload(self):
        # The measurement is an optimisation. If ffprobe cannot read the packets
        # of a file whose streams it just read happily, the upload still has to be
        # accepted on its metadata — an optimisation that can reject work is worse
        # than no optimisation.
        streams = json.dumps({"streams": [
            {"codec_type": "video", "r_frame_rate": "25/1", "avg_frame_rate": "25/1",
             "width": 640, "height": 480}],
            "format": {"duration": "4"}})
        with patch("clips.extraction._command",
                   side_effect=[streams, ValueError("ffprobe fell over")]):
            facts = extraction.probe(Path("awkward.mov"))
        self.assertEqual("25/1", facts["rate"])
        self.assertIsNone(facts["measured_fps"])

    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"])


@override_settings(BLOB_ROOT=BLOB_DIR)
class OwnershipTests(TestCase):
    """Anyone with the link reads; the owner and the editors write."""

    def setUp(self):
        from django.contrib.auth import get_user_model
        User = get_user_model()
        self.ann = User.objects.create_user("ann", password="password1")
        self.bob = User.objects.create_user("bob", password="password1")
        self.project = Project.objects.create(name="ann's", owner=self.ann)

    def write(self):
        return self.client.put(f"/api/projects/{self.project.id}",
                               data=json.dumps({"name": "renamed", "clips": []}),
                               content_type="application/json")

    def test_anyone_with_the_link_can_read_and_nobody_else_can_write(self):
        loaded = self.client.get(f"/api/projects/{self.project.id}").json()
        self.assertEqual(("ann", False), (loaded["owner"], loaded["can_edit"]))
        self.assertEqual(403, self.write().status_code)
        self.client.login(username="bob", password="password1")
        self.assertEqual(403, self.write().status_code)

    def test_the_owner_names_an_editor_who_can_then_write(self):
        self.client.login(username="bob", password="password1")
        self.assertEqual(403, self.client.post(
            f"/api/projects/{self.project.id}/editors", data=json.dumps({"username": "bob"}),
            content_type="application/json").status_code)
        self.client.login(username="ann", password="password1")
        self.assertEqual(200, self.write().status_code)
        self.assertEqual(["bob"], self.client.post(
            f"/api/projects/{self.project.id}/editors", data=json.dumps({"username": "bob"}),
            content_type="application/json").json()["editors"])
        self.client.login(username="bob", password="password1")
        self.assertTrue(self.client.get(f"/api/projects/{self.project.id}").json()["can_edit"])
        self.assertEqual(200, self.write().status_code)
        self.client.login(username="ann", password="password1")
        self.client.delete(f"/api/projects/{self.project.id}/editors/bob")
        self.client.login(username="bob", password="password1")
        self.assertEqual(403, self.write().status_code)

    def test_a_project_is_made_by_somebody_signed_in_and_is_theirs(self):
        self.assertEqual(403, self.client.post("/api/projects", data=json.dumps({"name": "x"}),
                                               content_type="application/json").status_code)
        self.client.post("/api/signup", data=json.dumps(
            {"username": "cat", "password": "password1"}), content_type="application/json")
        self.assertEqual("cat", self.client.get("/api/me").json()["username"])
        mine = self.client.post("/api/projects", data=json.dumps({"name": "y"}),
                                content_type="application/json").json()
        self.assertEqual(("cat", True), (mine["owner"], mine["can_edit"]))
        listed = {p["name"] for p in self.client.get("/api/projects").json()["projects"]}
        self.assertEqual({"y"}, listed)
        self.client.logout()
        self.assertEqual([], self.client.get("/api/projects").json()["projects"])

    def test_a_project_has_an_address(self):
        response = self.client.get(f"/p/{self.project.id}")
        self.assertEqual(200, response.status_code)
        # The slug is the name, for people; the id is what finds it.
        self.assertEqual(200, self.client.get(f"/p/{self.project.id}/anything-at-all").status_code)
        self.assertContains(response, 'id="app"')


class SocketTests(TestCase):
    """A committed write reaches everyone in the room; presence is stamped."""

    def test_a_save_is_broadcast_to_the_room(self):
        from asgiref.sync import async_to_sync, sync_to_async
        from channels.testing import WebsocketCommunicator
        from clips.consumers import broadcast
        from server.asgi import application

        from django.contrib.auth import get_user_model
        project = Project.objects.create(
            name="shared", owner=get_user_model().objects.create_user("host"))

        async def scenario():
            peer = WebsocketCommunicator(application, f"/ws/projects/{project.id}",
                                         headers=[(b"origin", b"http://localhost")])
            connected, _ = await peer.connect()
            self.assertTrue(connected)
            self.assertEqual("welcome", (await peer.receive_json_from())["kind"])
            self.assertEqual([], (await peer.receive_json_from())["peers"])
            self.assertEqual("join", (await peer.receive_json_from())["kind"])
            # The server stamps who sent it; a claimed name is overwritten.
            await peer.send_json_to({"kind": "state", "frame": 12, "user": "forged"})
            state = await peer.receive_json_from()
            self.assertEqual((12, None), (state["frame"], state["user"]))
            await sync_to_async(broadcast)(project.id, {"seq": 1, "clips": []})
            delta = await peer.receive_json_from()
            self.assertEqual(("delta", 1), (delta["kind"], delta["seq"]))
            await peer.disconnect()

        async_to_sync(scenario)()