lane G M17: memory capsule — zero-backend shareable bundle + Range-capable serve.py

build_capsule copies frontend/dist (clear error if missing), copies referenced media,
bakes every read-only API response extension-less at the live API's relative paths
(manifest with capsule:true + has_capture:false, events, anchors, poses, beats,
pointcloud, splat, audio via the same lazy extract_audio_hq cache, plus M16 saved
paths — listing baked as api/paths/index.html so per-id files can share the prefix),
injects window.__F4D_API_BASE__="" ahead of the app script, and ships a stdlib
serve.py (206 Range slicing incl. suffix ranges + 416, JSON content types for the
extension-less api/* files). Optional artifacts degrade by omission.

test_capsule.py: bundle structure, baked-JSON deep-equal vs the live route functions,
injection position, degrade/overwrite-guard/no-dist errors, CLI dispatch (CR-4), and
serve.py exercised over real HTTP in a subprocess (200/206/416/404, content types,
mp4+m4a magic bytes). Suite: 169 passed.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
type-two 2026-07-17 17:48:22 +10:00
parent b37f1fab0a
commit feb5588a1b
2 changed files with 628 additions and 4 deletions

View File

@ -1,4 +1,4 @@
"""Memory capsule: zero-backend shareable export (spec M17, lane G). STUB — lane G fills. """Memory capsule: zero-backend shareable export (spec M17, lane G).
Contract: :func:`build_capsule` writes a self-contained bundle to ``out_dir`` (default Contract: :func:`build_capsule` writes a self-contained bundle to ``out_dir`` (default
``config.CAPSULE_DIR``): ``config.CAPSULE_DIR``):
@ -16,13 +16,300 @@ Contract: :func:`build_capsule` writes a self-contained bundle to ``out_dir`` (d
Skip what doesn't exist (no splat -> no ``api/splat``) — degrade, don't block. Skip what doesn't exist (no splat -> no ``api/splat``) — degrade, don't block.
Returns a summary dict (bundle path, file count, bytes) for the CLI log. Returns a summary dict (bundle path, file count, bytes) for the CLI log.
Layout mirror (why paths look the way they do): the frontend resolves every URL as
``API_BASE + path`` with API_BASE == "" in a capsule, so requests hit the bundle root:
``fetch("/api/manifest")`` (no extension bake the file extension-less; serve.py maps
extension-less ``api/*`` files to a JSON content type), ``video.src = "/media/<filename>"``
(manifest ``url`` fields are ``/media/{filename}`` media files are copied under
``media/``), PLY loader hits ``/api/pointcloud``, the WebAudio clock fetches
``/api/audio/{id}``. Saved camera paths (M16) are read-only in a capsule, so
``api/paths`` + ``api/paths/{id}`` are baked too the load dropdown keeps working.
""" """
from __future__ import annotations from __future__ import annotations
import json
import logging
import shutil
from pathlib import Path from pathlib import Path
from festival4d import config, db
def build_capsule(out_dir: str | Path | None = None) -> dict: log = logging.getLogger("festival4d.capsule")
"""Build the static shareable bundle (see module doc)."""
raise NotImplementedError("memory capsule (lane G / M17)") # Injected verbatim (phase-5 contract #5): a runtime override that beats the build-time
# VITE_API_BASE, turning the same dist bundle zero-backend. Must land in index.html before
# the app module reads it (modules are deferred, but we inject ahead of the first <script>
# anyway so execution order is obvious).
_API_BASE_SNIPPET = '<script>window.__F4D_API_BASE__=""</script>'
# ---------------------------------------------------------------------------
# serve.py shipped in the bundle root (stdlib-only, Range-capable — pitfall #4)
# ---------------------------------------------------------------------------
SERVE_PY = r'''#!/usr/bin/env python3
"""Serve this Festival 4D memory capsule locally.
Plain `python -m http.server` cannot answer HTTP Range requests, and <video>
seeking needs 206 partial responses so the capsule ships this tiny stdlib
server instead. No dependencies beyond Python 3.8+.
python serve.py [--port 8080] [--host 127.0.0.1]
Then open http://127.0.0.1:8080/ in a browser.
"""
import argparse
import os
import re
from functools import partial
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
ROOT = os.path.dirname(os.path.abspath(__file__))
_RANGE = re.compile(r"bytes=(\d*)-(\d*)$")
class _Slice:
"""File wrapper that reads at most `length` bytes (the body of a 206)."""
def __init__(self, f, length):
self._f = f
self._left = length
def read(self, n=-1):
if self._left <= 0:
return b""
if n < 0 or n > self._left:
n = self._left
data = self._f.read(n)
self._left -= len(data)
return data
def close(self):
self._f.close()
class RangeHandler(SimpleHTTPRequestHandler):
protocol_version = "HTTP/1.1"
def guess_type(self, path):
# Baked API responses are extension-less files at the live API's relative
# paths; give them the content types the live server would have used.
url = self.path.split("?", 1)[0].split("#", 1)[0].lstrip("/")
if "." not in url.rsplit("/", 1)[-1]:
if url.startswith("api/audio/"):
return "audio/mp4"
if url in ("api/pointcloud", "api/splat"):
return "application/octet-stream"
if url.startswith("api/"):
return "application/json"
return super().guess_type(path)
def send_head(self):
path = self.translate_path(self.path)
if os.path.isdir(path):
# Directory -> index.html (the SPA entry), stock behavior.
return super().send_head()
try:
f = open(path, "rb")
except OSError:
self.send_error(404, "File not found")
return None
try:
size = os.fstat(f.fileno()).st_size
start, end, status = 0, size - 1, 200
m = _RANGE.match(self.headers.get("Range", "").strip())
if m and (m.group(1) or m.group(2)):
if m.group(1):
start = int(m.group(1))
if m.group(2):
end = min(int(m.group(2)), size - 1)
else: # suffix form: bytes=-N (the final N bytes)
start = max(size - int(m.group(2)), 0)
if start >= size:
self.send_response(416)
self.send_header("Content-Range", "bytes */%d" % size)
self.send_header("Content-Length", "0")
self.end_headers()
f.close()
return None
status = 206
self.send_response(status)
self.send_header("Content-Type", self.guess_type(path))
self.send_header("Accept-Ranges", "bytes")
self.send_header("Content-Length", str(end - start + 1))
if status == 206:
self.send_header("Content-Range", "bytes %d-%d/%d" % (start, end, size))
self.end_headers()
f.seek(start)
return _Slice(f, end - start + 1)
except Exception:
f.close()
raise
def main():
ap = argparse.ArgumentParser(description="Serve this Festival 4D capsule (Range-capable).")
ap.add_argument("--host", default="127.0.0.1")
ap.add_argument("--port", type=int, default=8080)
args = ap.parse_args()
handler = partial(RangeHandler, directory=ROOT)
httpd = ThreadingHTTPServer((args.host, args.port), handler)
print("Festival 4D capsule -> http://%s:%d/" % (args.host, args.port))
try:
httpd.serve_forever()
except KeyboardInterrupt:
pass
if __name__ == "__main__":
main()
'''
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _bake_json(path: Path, obj) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(obj, indent=2))
def _inject_api_base(index_html: Path) -> None:
"""Insert the runtime API-base override into the copied index.html."""
html = index_html.read_text()
if "__F4D_API_BASE__" in html:
return # already injected (rebuild over an old bundle copy)
i = html.find("<script")
if i == -1:
i = html.find("</head>")
if i != -1:
html = html[:i] + _API_BASE_SNIPPET + "\n " + html[i:]
else: # no <script> and no </head> — degenerate page; prepend
html = _API_BASE_SNIPPET + "\n" + html
index_html.write_text(html)
def _prepare_out_dir(out: Path) -> None:
"""Refuse to clobber a directory that isn't (or wasn't) a capsule."""
if not out.exists():
return
contents = list(out.iterdir())
is_previous_capsule = (out / "serve.py").exists() or (out / "index.html").exists()
if contents and not is_previous_capsule:
raise RuntimeError(
f"output dir {out} exists and does not look like a previous capsule — "
"refusing to overwrite; pass an empty/new --out"
)
shutil.rmtree(out)
def _summary(out: Path) -> dict:
files = [p for p in out.rglob("*") if p.is_file()]
return {
"bundle": str(out),
"files": len(files),
"bytes": sum(p.stat().st_size for p in files),
}
# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------
def build_capsule(out_dir: str | Path | None = None,
dist_dir: str | Path | None = None) -> dict:
"""Build the static shareable bundle (see module doc).
``dist_dir`` overrides the frontend build location (default
``<repo>/frontend/dist``) used by tests to point at a fabricated dist.
"""
# The API module is the single source of truth for response shapes — its route
# functions are plain callables returning the exact dicts the live server sends,
# so baking through them can never drift from the frozen contract.
from festival4d import api, ingest
dist = Path(dist_dir) if dist_dir is not None else config.REPO_ROOT / "frontend" / "dist"
if not (dist / "index.html").exists():
raise RuntimeError(
f"frontend build not found at {dist} — run `cd frontend && npm run build` first"
)
out = Path(out_dir) if out_dir is not None else config.CAPSULE_DIR
_prepare_out_dir(out)
db.init_engine()
db.init_db()
videos = db.get_videos()
# 1. Frontend bundle + runtime API-base override.
shutil.copytree(dist, out)
_inject_api_base(out / "index.html")
# 2. Media (only the files the manifest references — RAW_DIR may hold strays).
media_dir = out / "media"
media_dir.mkdir(parents=True, exist_ok=True)
for v in videos:
src = config.RAW_DIR / v.filename
if src.exists():
shutil.copy2(src, media_dir / v.filename)
else:
log.warning("capsule: source video missing, skipped: %s", src)
# 3. Baked read-only API responses, at the SAME relative paths the live API serves
# (extension-less on purpose — the frontend fetches e.g. `${API_BASE}/api/manifest`).
api_dir = out / "api"
manifest = api.manifest()
manifest["capsule"] = True # frontend hides all write UI (contract #5)
manifest["has_capture"] = False # capture routes never exist in a static bundle
_bake_json(api_dir / "manifest", manifest)
_bake_json(api_dir / "events", api.get_events())
_bake_json(api_dir / "anchors", api.get_anchors())
for v in videos:
_bake_json(api_dir / "videos" / str(v.id) / "poses", api.video_poses(v.id))
# Saved camera paths (M16) — read-only in a capsule, so the load dropdown still works.
# A filesystem can't hold both a FILE `api/paths` and per-id files under `api/paths/`,
# so the listing is baked as the directory's index.html: GET /api/paths gets the stock
# 301 -> /api/paths/ -> index.html (fetch follows redirects), and serve.py's URL-based
# content-type override still labels it application/json.
_bake_json(api_dir / "paths" / "index.html", api.get_paths())
for p in db.get_paths():
_bake_json(api_dir / "paths" / str(p.id), api.get_path(p.id))
# Optional artifacts: bake when present, silently omit when not (the frontend already
# handles the 404s serve.py will answer for the missing files — degrade, don't block).
if config.BEATS_JSON.exists():
_bake_json(api_dir / "beats", json.loads(config.BEATS_JSON.read_text()))
if config.POINTS_PLY.exists():
api_dir.mkdir(parents=True, exist_ok=True)
shutil.copy2(config.POINTS_PLY, api_dir / "pointcloud")
if config.SPLAT_PLY.exists():
api_dir.mkdir(parents=True, exist_ok=True)
shutil.copy2(config.SPLAT_PLY, api_dir / "splat")
# 4. Soundtracks for the WebAudio master clock: reuse the live server's lazy cache,
# extracting on demand exactly like GET /api/audio/{id} would have.
import subprocess
for v in videos:
src = config.RAW_DIR / v.filename
cached = config.AUDIO_HQ_DIR / f"{v.id}.m4a"
if src.exists() and (not cached.exists() or cached.stat().st_mtime < src.stat().st_mtime):
try:
ingest.extract_audio_hq(src, cached)
except (subprocess.CalledProcessError, RuntimeError, OSError) as exc:
log.warning("capsule: no extractable audio for video %s: %s", v.id, exc)
if cached.exists():
audio_out = api_dir / "audio" / str(v.id)
audio_out.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(cached, audio_out)
# 5. The Range-capable server (pitfall #4).
serve_py = out / "serve.py"
serve_py.write_text(SERVE_PY)
serve_py.chmod(serve_py.stat().st_mode | 0o111)
summary = _summary(out)
log.info("capsule: baked %(files)d files (%(bytes)d bytes) at %(bundle)s", summary)
return summary

View File

@ -0,0 +1,337 @@
"""M17 memory capsule (lane G): bundle structure, baked-JSON validity, serve.py Range.
Builds a real capsule from the synthetic project (temp data dir per conftest) against a
fabricated frontend dist (spec allows this running npm inside tests would be neither
hermetic nor fast), then boots the bundled ``serve.py`` in a subprocess and exercises it
over HTTP: 200s, JSON content types on the extension-less baked API files, and the 206
Range behavior video seeking depends on (phase-5 pitfall #4).
"""
from __future__ import annotations
import json
import shutil
import socket
import subprocess
import sys
import time
import urllib.error
import urllib.request
import pytest
pytestmark = pytest.mark.skipif(
shutil.which("ffmpeg") is None or shutil.which("ffprobe") is None,
reason="ffmpeg/ffprobe required to build the synthetic capsule fixture",
)
CAMPATH = {
"version": 1,
"keyframes": [
{"t_global": 0.2, "pos": [0.0, 1.0, 5.0], "quat": [0.0, 0.0, 0.0, 1.0], "fov": 50.0},
{"t_global": 0.8, "pos": [2.0, 1.0, 4.0], "quat": [0.0, 0.5, 0.0, 0.8660254], "fov": 42.0},
],
}
BEATS = {"tempo_bpm": 120.0, "beats_s": [0.25, 0.75], "onsets": [],
"generated_by": "festival4d.audio_features"}
SPLAT_BYTES = b"ply\nfake-splat-for-tests\n"
API_SNIPPET = '<script>window.__F4D_API_BASE__=""</script>'
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture(scope="module")
def project():
"""Synthetic project + every optional artifact seeded (beats, splat, a saved path)."""
from festival4d import config, db, synthetic
synthetic.build(duration_s=1.0, run_ffmpeg=True) # into the temp data dir (conftest)
db.init_engine()
db.init_db()
config.BEATS_JSON.parent.mkdir(parents=True, exist_ok=True)
config.BEATS_JSON.write_text(json.dumps(BEATS))
config.SPLAT_PLY.write_bytes(SPLAT_BYTES)
saved = db.add_path("capsule sweep", json.dumps(CAMPATH))
yield {"path_id": saved.id}
# Leave the shared session data dir as later test modules expect to find it.
config.BEATS_JSON.unlink(missing_ok=True)
config.SPLAT_PLY.unlink(missing_ok=True)
db.delete_path(saved.id)
def _make_dist(root):
"""A minimal Vite-shaped dist: index.html with a module <script> + one asset."""
root.mkdir(parents=True, exist_ok=True)
(root / "assets").mkdir(exist_ok=True)
(root / "assets" / "index.js").write_text("console.log('festival4d test dist');\n")
(root / "index.html").write_text(
"<!doctype html>\n<html>\n <head>\n <meta charset=\"UTF-8\" />\n"
" <title>Festival 4D</title>\n"
" <script type=\"module\" crossorigin src=\"/assets/index.js\"></script>\n"
" </head>\n <body><div id=\"app\"></div></body>\n</html>\n"
)
return root
@pytest.fixture(scope="module")
def fake_dist(tmp_path_factory):
return _make_dist(tmp_path_factory.mktemp("frontend") / "dist")
@pytest.fixture(scope="module")
def bundle(project, fake_dist, tmp_path_factory):
from festival4d import capsule
out = tmp_path_factory.mktemp("capsule") / "bundle"
summary = capsule.build_capsule(out_dir=out, dist_dir=fake_dist)
return out, summary
@pytest.fixture(scope="module")
def server(bundle):
"""The bundle's own serve.py, running as a real subprocess in the bundle root."""
out, _ = bundle
with socket.socket() as s:
s.bind(("127.0.0.1", 0))
port = s.getsockname()[1]
proc = subprocess.Popen(
[sys.executable, "serve.py", "--port", str(port)],
cwd=out, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
)
base = f"http://127.0.0.1:{port}"
try:
deadline = time.monotonic() + 15
while True:
try:
urllib.request.urlopen(f"{base}/api/manifest", timeout=1).read()
break
except OSError:
if proc.poll() is not None or time.monotonic() > deadline:
raise RuntimeError(
f"serve.py did not come up: {proc.stdout.read().decode(errors='replace')}"
)
time.sleep(0.1)
yield base
finally:
proc.terminate()
proc.wait(timeout=5)
def _get(url, headers=None):
"""GET returning (status, headers, body); headers stay a case-insensitive Message."""
req = urllib.request.Request(url, headers=headers or {})
try:
with urllib.request.urlopen(req, timeout=5) as resp:
return resp.status, resp.headers, resp.read()
except urllib.error.HTTPError as err:
return err.code, err.headers, err.read()
# ---------------------------------------------------------------------------
# Bundle structure
# ---------------------------------------------------------------------------
def test_bundle_structure(bundle):
from festival4d import db
out, summary = bundle
for rel in ("index.html", "serve.py", "assets/index.js",
"api/manifest", "api/events", "api/anchors",
"api/beats", "api/pointcloud", "api/splat",
# the paths listing lives at paths/index.html so per-id files can
# coexist under the same URL prefix (see capsule.py)
"api/paths/index.html"):
assert (out / rel).is_file(), f"missing {rel}"
for v in db.get_videos():
assert (out / "media" / v.filename).is_file()
assert (out / "api" / "videos" / str(v.id) / "poses").is_file()
assert (out / "api" / "audio" / str(v.id)).is_file()
assert summary["bundle"] == str(out)
assert summary["files"] >= 10
assert summary["bytes"] > 0
def test_baked_manifest_is_capsule_true_and_urls_resolve(bundle):
out, _ = bundle
manifest = json.loads((out / "api" / "manifest").read_text())
assert manifest["capsule"] is True
assert manifest["has_capture"] is False # capture routes never exist in a static bundle
assert manifest["has_poses"] is True and manifest["has_splat"] is True
assert manifest["t_global_max"] > 0
assert len(manifest["videos"]) >= 2
for v in manifest["videos"]:
# Every manifest URL must resolve inside the bundle (API_BASE == "" ⇒ root-relative).
assert v["url"].startswith("/media/")
assert (out / v["url"].lstrip("/")).is_file()
def test_baked_json_matches_live_api_shapes(bundle, project):
from festival4d import api, db
out, _ = bundle
manifest = json.loads((out / "api" / "manifest").read_text())
live = api.manifest()
assert set(manifest) == set(live) # key set identical to the live route
assert json.loads((out / "api" / "events").read_text()) == api.get_events()
assert json.loads((out / "api" / "anchors").read_text()) == api.get_anchors()
for v in db.get_videos():
baked = json.loads((out / "api" / "videos" / str(v.id) / "poses").read_text())
assert baked == api.video_poses(v.id)
assert len(baked) > 0
assert json.loads((out / "api" / "beats").read_text()) == BEATS
assert (out / "api" / "splat").read_bytes() == SPLAT_BYTES
assert (out / "api" / "pointcloud").read_bytes()[:3] == b"ply"
# Saved camera paths (M16) baked read-only so the load dropdown works zero-backend.
listing = json.loads((out / "api" / "paths" / "index.html").read_text())
assert [p["name"] for p in listing] == ["capsule sweep"]
pid = project["path_id"]
assert json.loads((out / "api" / "paths" / str(pid)).read_text())["json"] == CAMPATH
def test_index_html_injects_api_base_before_app_script(bundle):
out, _ = bundle
html = (out / "index.html").read_text()
assert html.count(API_SNIPPET) == 1
assert html.index(API_SNIPPET) < html.index('type="module"')
def test_optional_artifacts_omitted_when_absent(project, fake_dist, tmp_path):
"""Degrade, don't block: no beats/splat ⇒ those files are simply not baked."""
from festival4d import capsule, config
beats, splat = config.BEATS_JSON.read_text(), config.SPLAT_PLY.read_bytes()
config.BEATS_JSON.unlink()
config.SPLAT_PLY.unlink()
try:
out = tmp_path / "bare"
capsule.build_capsule(out_dir=out, dist_dir=fake_dist)
assert not (out / "api" / "beats").exists()
assert not (out / "api" / "splat").exists()
manifest = json.loads((out / "api" / "manifest").read_text())
assert manifest["has_splat"] is False
assert manifest["capsule"] is True
finally:
config.BEATS_JSON.write_text(beats)
config.SPLAT_PLY.write_bytes(splat)
def test_missing_dist_errors_clearly(tmp_path):
from festival4d import capsule
with pytest.raises(RuntimeError, match="npm run build"):
capsule.build_capsule(out_dir=tmp_path / "never", dist_dir=tmp_path / "no-dist")
assert not (tmp_path / "never").exists()
def test_rebuild_overwrites_previous_capsule_but_not_strangers(bundle, fake_dist, project, tmp_path):
from festival4d import capsule
# A non-capsule, non-empty directory is refused (don't eat someone's folder).
stranger = tmp_path / "stranger"
stranger.mkdir()
(stranger / "precious.txt").write_text("do not delete")
with pytest.raises(RuntimeError, match="refusing"):
capsule.build_capsule(out_dir=stranger, dist_dir=fake_dist)
assert (stranger / "precious.txt").exists()
# Rebuilding over a previous capsule works and never double-injects the snippet.
out, _ = bundle
capsule.build_capsule(out_dir=out, dist_dir=fake_dist)
assert (out / "index.html").read_text().count(API_SNIPPET) == 1
def test_cli_capsule_dispatch(project, fake_dist, tmp_path, monkeypatch):
"""`python -m festival4d capsule` (CR-4: the stub exit-2 contract is retired)."""
from festival4d import cli, config
# The CLI has no --dist flag; point the repo-root lookup at a fabricated tree.
fake_root = tmp_path / "repo"
_make_dist(fake_root / "frontend" / "dist")
monkeypatch.setattr(config, "REPO_ROOT", fake_root)
out = tmp_path / "cli-bundle"
assert cli.main(["capsule", "--out", str(out)]) == 0
assert (out / "serve.py").is_file()
assert json.loads((out / "api" / "manifest").read_text())["capsule"] is True
# And the no-dist error path is a clear message, not a stub exit.
monkeypatch.setattr(config, "REPO_ROOT", tmp_path / "empty-repo")
with pytest.raises(RuntimeError, match="npm run build"):
cli.main(["capsule", "--out", str(tmp_path / "nope")])
# ---------------------------------------------------------------------------
# serve.py over real HTTP
# ---------------------------------------------------------------------------
def test_serve_index_and_json_content_types(server):
status, headers, body = _get(f"{server}/")
assert status == 200
assert headers["Content-Type"].startswith("text/html")
assert API_SNIPPET.encode() in body
status, headers, body = _get(f"{server}/api/manifest")
assert status == 200
assert headers["Content-Type"] == "application/json" # extension-less baked file
assert json.loads(body)["capsule"] is True
for rel, ctype in (("api/events", "application/json"),
("api/paths", "application/json"),
("api/pointcloud", "application/octet-stream"),
("api/splat", "application/octet-stream")):
status, headers, _ = _get(f"{server}/{rel}")
assert (status, headers["Content-Type"]) == (200, ctype), rel
def test_serve_video_range_206(server, bundle):
out, _ = bundle
manifest = json.loads((out / "api" / "manifest").read_text())
url = manifest["videos"][0]["url"] # "/media/<filename>"
size = (out / url.lstrip("/")).stat().st_size
# Plain GET: 200, full body, and Range support advertised (browsers probe this).
status, headers, body = _get(f"{server}{url}")
assert status == 200
assert headers["Accept-Ranges"] == "bytes"
assert int(headers["Content-Length"]) == size == len(body)
# bytes=0-100 ⇒ 206 with exactly 101 bytes and a correct Content-Range.
status, headers, body = _get(f"{server}{url}", {"Range": "bytes=0-100"})
assert status == 206
assert headers["Content-Range"] == f"bytes 0-100/{size}"
assert len(body) == 101
assert body[4:8] == b"ftyp" # mp4 magic — it's really the video's first bytes
# Open-ended seek (bytes=N-) ⇒ the tail from N.
mid = size // 2
status, headers, body = _get(f"{server}{url}", {"Range": f"bytes={mid}-"})
assert status == 206
assert headers["Content-Range"] == f"bytes {mid}-{size - 1}/{size}"
assert len(body) == size - mid
# Suffix form (bytes=-N) ⇒ the final N bytes (mp4 moov probing does this).
status, headers, body = _get(f"{server}{url}", {"Range": "bytes=-50"})
assert status == 206
assert headers["Content-Range"] == f"bytes {size - 50}-{size - 1}/{size}"
assert len(body) == 50
# Start past EOF ⇒ 416 with the total size.
status, headers, _ = _get(f"{server}{url}", {"Range": f"bytes={size + 10}-"})
assert status == 416
assert headers["Content-Range"] == f"bytes */{size}"
def test_serve_audio_bytes_and_range(server, bundle):
status, headers, body = _get(f"{server}/api/audio/1")
assert status == 200
assert headers["Content-Type"] == "audio/mp4"
assert body[4:8] == b"ftyp" # m4a container magic
# The WebAudio clock fetches whole files, but Range must still work on audio.
status, headers, part = _get(f"{server}/api/audio/1", {"Range": "bytes=0-15"})
assert status == 206 and part == body[:16]
def test_serve_missing_file_404(server):
assert _get(f"{server}/api/nope")[0] == 404
assert _get(f"{server}/media/ghost.mp4")[0] == 404