From e7d2eb8cf6f3cd25f1f316ee542f809800621eca Mon Sep 17 00:00:00 2001 From: npeter83 Date: Mon, 5 Oct 2026 21:05:22 +0200 Subject: [PATCH] test(downloads): what a share link serves passes the phone-safe rule MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A link serves an asset once it is ready, so the contract is pinned on both sides. The code: every function in app/ that marks something ready must run the rule (`_ensure_phone_safe`, or the upload's own `playable.plan`) — an AST scan names any that does not, and fails blind if it stops finding the four paths known today (download, editor clip, upload conversion, upload kept as is). The bytes: the files an iPhone + iPad refused (HEVC tagged hev1, AV1, Matroska, MP3 inside an .mp4) and the ones other browsers cannot be trusted with (VP9 + Opus, 10-bit H.264, Opus in Ogg) go through the upload route, the worker's conversion and the public file route; what the link hands out must be served as mp4 / m4a / mp3 and pass the rule. Seen red: a planted ready-marking function in service.py is named; with the conversion step short-circuited, all seven files fail (on the rule, or on being served as quicktime / webm / ogg). --- backend/tests/test_phone_safe_contract.py | 163 ++++++++++++++++++++++ 1 file changed, 163 insertions(+) create mode 100644 backend/tests/test_phone_safe_contract.py diff --git a/backend/tests/test_phone_safe_contract.py b/backend/tests/test_phone_safe_contract.py new file mode 100644 index 00000000..50505f73 --- /dev/null +++ b/backend/tests/test_phone_safe_contract.py @@ -0,0 +1,163 @@ +"""The share-link contract: every media file a /watch link can serve passes the phone-safe rule +(`downloads.playable`). + +Two halves. The structural one reads the code: a link serves an asset once it is `ready`, so every +function that marks one ready must run the rule — a new path to `ready` that skips it is named here, +before a friend's phone meets the file. The end-to-end one sends the files measured broken on an +iPhone + iPad (and the ones browsers other than Safari cannot be trusted with) through upload, the +worker's conversion and the public file route, and judges the bytes the link actually hands out. +""" + +import ast +import shutil +import subprocess +from pathlib import Path + +import pytest + +from app.config import settings +from app.downloads import pipeline, playable, probe +from app.models import DownloadJob, MediaAsset + +APP = Path(__file__).resolve().parents[1] / "app" + +# The calls that ARE the rule: the worker's step (download, clip, upload conversion) and the upload's +# judgement of a file it keeps as it came. +_RULE_CALLS = {"_ensure_phone_safe", "plan"} + + +def _marks_ready(node: ast.AST) -> bool: + """`x.status = "ready"`, `status="ready"`, or `"…" if … else "ready"` handed to either.""" + + def is_ready(v: ast.AST | None) -> bool: + if isinstance(v, ast.Constant): + return v.value == "ready" + if isinstance(v, ast.IfExp): + return is_ready(v.body) or is_ready(v.orelse) + return False + + if isinstance(node, ast.Assign): + return is_ready(node.value) and any( + isinstance(t, ast.Attribute) and t.attr == "status" for t in node.targets + ) + return isinstance(node, ast.keyword) and node.arg == "status" and is_ready(node.value) + + +def _calls(fn: ast.AST) -> set[str]: + names = set() + for n in ast.walk(fn): + if isinstance(n, ast.Call): + f = n.func + names.add(f.attr if isinstance(f, ast.Attribute) else getattr(f, "id", "")) + return names + + +def ready_sites(source: str) -> list[tuple[str, int, bool]]: + """(function, line, runs the rule) for every place `source` marks something ready.""" + sites = [] + for fn in ast.walk(ast.parse(source)): + if not isinstance(fn, ast.FunctionDef | ast.AsyncFunctionDef): + continue + runs_rule = bool(_calls(fn) & _RULE_CALLS) + for n in ast.walk(fn): + if _marks_ready(n): + sites.append((fn.name, n.lineno, runs_rule)) + return sites + + +def test_every_path_to_ready_runs_the_rule(): + found, skipped = [], [] + for path in sorted(APP.rglob("*.py")): + for name, line, runs_rule in ready_sites(path.read_text(encoding="utf-8")): + where = f"{path.relative_to(APP.parent).as_posix()}:{line} {name}" + found.append(where) + if not runs_rule: + skipped.append(where) + # The paths known today: a download, an editor clip, an upload's conversion, an upload kept as is. + # Fewer means the scan went blind (a rename, a new status spelling) — not that a path is gone. + names = {w.rsplit(" ", 1)[1] for w in found} + assert {"_download", "_process_edit", "_convert_upload", "ingest_upload"} <= names, found + assert skipped == [], f"marks an asset ready without the phone-safe rule: {skipped}" + + +def test_the_scan_names_a_path_that_skips_the_rule(): + planted = 'def restore(asset):\n asset.status = "ready"\n' + guarded = ( + 'def restore(asset, p):\n _ensure_phone_safe(1, p, p, "x")\n asset.status = "ready"\n' + ) + assert ready_sites(planted) == [("restore", 2, False)] + assert ready_sites(guarded) == [("restore", 3, True)] + assert ( + ready_sites('def f(a, x):\n a.update(status="pending" if x else "ready")\n')[0][2] + is False + ) + + +# --- end to end: upload → worker → the bytes a link serves --------------------------------------- + +_V = ["-f", "lavfi", "-i", "testsrc2=size=160x120:rate=10:duration=1"] +_A = ["-f", "lavfi", "-i", "sine=duration=1"] + +# name -> (ffmpeg args after the inputs, filename the user uploads) +_FILES = { + "hevc tagged hev1 (iOS: broken)": ( + [*_V, *_A, "-c:v", "libx265", "-tag:v", "hev1", "-c:a", "aac"], + "phone.mov", + ), + "av1 (iOS: broken)": ([*_V, *_A, "-c:v", "libsvtav1", "-c:a", "aac"], "clip.mp4"), + "h264 in matroska (iOS: broken)": ([*_V, *_A, "-c:v", "libx264", "-c:a", "aac"], "clip.mkv"), + "mp3 inside an mp4 (iOS: broken)": ([*_A, "-c:a", "libmp3lame", "-f", "mp4"], "song.mp4"), + "vp9 + opus webm": ([*_V, *_A, "-c:v", "libvpx-vp9", "-c:a", "libopus"], "clip.webm"), + "10-bit h264": ( + [*_V, *_A, "-c:v", "libx264", "-pix_fmt", "yuv420p10le", "-c:a", "aac"], + "clip.mp4", + ), + "opus in ogg": ([*_A, "-c:a", "libopus"], "voice.ogg"), +} + + +@pytest.fixture +def media_root(tmp_path, monkeypatch): + root = tmp_path / "media" + root.mkdir() + monkeypatch.setattr(settings, "download_root", str(root)) + return root + + +@pytest.mark.skipif(shutil.which("ffmpeg") is None, reason="ffmpeg not available") +@pytest.mark.parametrize("name", _FILES) +def test_what_a_link_serves_passes_the_rule( + name, db, as_user, make_user, client, media_root, tmp_path, upload_headers +): + args, filename = _FILES[name] + src = tmp_path / f"src{Path(filename).suffix}" + subprocess.run(["ffmpeg", "-y", "-loglevel", "error", *args, str(src)], check=True) + + owner = as_user(make_user()) + r = owner.post( + "/api/downloads/upload", content=src.read_bytes(), headers=upload_headers(filename) + ) + assert r.status_code == 200, r.text + job_id = r.json()["id"] + assert r.json()["status"] == "queued" # every one of these fails the rule as it came + job = db.get(DownloadJob, job_id) + asset = db.get(MediaAsset, job.asset_id) + job.status, asset.status = "running", "downloading" # what the worker's claim does + db.commit() + pipeline._convert_upload(job.id, asset.id) + db.expire_all() + assert db.get(DownloadJob, job_id).status == "done" + + link = owner.post(f"/api/downloads/{job_id}/links", json={}) + assert link.status_code == 200, link.text + served = client.get(f"/api/public/watch/{link.json()['token']}/file") + assert served.status_code == 200 + got = tmp_path / "served.bin" + got.write_bytes(served.content) + data = probe.probe_json(got) + ext = {"video/mp4": "mp4", "audio/mp4": "m4a", "audio/mpeg": "mp3"}.get( + served.headers["content-type"].split(";")[0] + ) + assert ext is not None, served.headers["content-type"] # served as a type every phone plays + verdict = playable.plan(data, ext) + assert verdict is not None and not verdict.needs_change, (name, data["streams"])