"""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.downloads.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 ad-hoc 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", "_process_clip", "_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, "files_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/files/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"])