1120 lines
56 KiB
Python
1120 lines
56 KiB
Python
"""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(6, 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_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"])
|
|
|
|
|
|
@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)()
|