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 struct
import tempfile
import zlib
from pathlib import Path

from django.test import TestCase, override_settings

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

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)


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


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