siftlode/backend/tests/test_phone_safe_contract.py

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