siftlode/backend/tests/test_upload_conversion.py

"""An upload after ingest: what it costs its owner from the moment it lands, how the worker's
conversion names, checks and cleans up the file it makes, and what the owner and a share link see.

DB-backed. The conversion's ffmpeg step (`pipeline._ensure_phone_safe`) is stubbed to write a file of
a chosen size — these tests are about what happens around the encode; `test_upload.py` runs the real
one, and `test_playable.py` judges its bytes.
"""

import os
import time
from datetime import datetime, timedelta, timezone
from pathlib import Path

import pytest

from app import sysconfig
from app.config import settings
from app.downloads import errors as dlerrors
from app.downloads import gc as gcmod
from app.downloads import pipeline, quota, storage
from app.downloads import probe as probemod
from app.downloads import upload as uploadmod
from app.downloads.stated import StatedFailure
from app.models import DownloadJob, DownloadQuota, MediaAsset, Notification
from sqlalchemy import select, update
from sqlalchemy.exc import OperationalError
from sqlalchemy.orm import Session


@pytest.fixture
def _root(tmp_path, monkeypatch):
    monkeypatch.setattr(settings, "files_root", str(tmp_path))
    return tmp_path


def _probe(monkeypatch, vcodec="hevc"):
    """ffprobe's view of every file here: `vcodec` + AAC, 8-bit 4:2:0 — HEVC is queued for the
    worker, H.264 lands as it came."""
    data = {
        "streams": [
            {"codec_type": "video", "codec_name": vcodec, "pix_fmt": "yuv420p",
             "width": 640, "height": 360},
            {"codec_type": "audio", "codec_name": "aac"},
        ],
        "format": {"format_name": "mov,mp4", "duration": "6"},
    }  # fmt: skip
    monkeypatch.setattr(probemod, "probe_json", lambda _p: data)


def _ingest(db, root, user_id, name="clip.mov", size=10):
    src = root / "incoming.bin"
    src.write_bytes(b"o" * size)
    job = uploadmod.ingest_upload(db, user_id, filename=name, src_path=src, size=size)
    return job, db.get(MediaAsset, job.asset_id)


def _claim(db, job, asset):
    """What the worker's claim does before it converts."""
    job.status, asset.status = "running", "downloading"
    db.commit()


def _converts_to(monkeypatch, size, phases=None):
    """The encode writes a `size`-byte .mp4 into its staging dir (and notes the phase it reports)."""

    def _encode(job_id, produced, out_dir, phase, total_s=None, **_k):
        if phases is not None:
            phases.append(phase)
        out = Path(out_dir) / "converted.mp4"
        out.write_bytes(b"c" * size)
        return out

    monkeypatch.setattr(pipeline, "_ensure_phone_safe", _encode)


def _cap(db, user_id, max_bytes):
    db.add(
        DownloadQuota(
            user_id=user_id, max_bytes=max_bytes, max_concurrent=2, max_jobs=50, unlimited=False
        )
    )
    db.commit()


def _uploads_dir(root, user_id) -> Path:
    return root / ".uploads" / str(user_id)


# --- what it costs ---------------------------------------------------------------------------


def test_a_queued_upload_counts_against_its_owner_from_the_moment_it_lands(
    db, make_user, _root, monkeypatch
):
    _probe(monkeypatch)
    u = make_user()
    job, asset = _ingest(db, _root, u.id, size=10)
    assert (job.status, asset.status) == ("queued", "pending")
    assert quota.footprint(db, u.id) == 10  # its original is on disk already


def test_a_failed_conversion_still_counts_its_kept_original(db, make_user, _root, monkeypatch):
    _probe(monkeypatch)
    u = make_user()
    job, asset = _ingest(db, _root, u.id, size=10)
    job.status, asset.status = "error", "error"  # kept for Retry
    db.commit()
    assert storage.abs_path(str(_root), asset.rel_path).exists()
    assert quota.footprint(db, u.id) == 10


def test_a_conversion_that_outgrows_the_cap_is_refused_and_keeps_the_original(
    db, make_user, _root, monkeypatch
):
    _probe(monkeypatch)
    u = make_user()
    _cap(db, u.id, max_bytes=100)
    job, asset = _ingest(db, _root, u.id, size=10)
    original = storage.abs_path(str(_root), asset.rel_path)
    _claim(db, job, asset)
    _converts_to(monkeypatch, size=500)

    pipeline._convert_upload(job.id, asset.id)

    db.expire_all()
    job = db.get(DownloadJob, job.id)
    assert (job.status, job.error_code) == ("error", "quota_exceeded")
    # 90 of the 100 are still free: the converted file is too big, the storage is not full.
    assert job.error == dlerrors.CONVERTED_TOO_BIG_MESSAGE
    assert original.exists()
    assert [p.name for p in _uploads_dir(_root, u.id).iterdir()] == [original.name]


def test_a_conversion_that_shrinks_passes_an_owner_at_the_cap(db, make_user, _root, monkeypatch):
    _probe(monkeypatch)
    u = make_user()
    _cap(db, u.id, max_bytes=10)
    job, asset = _ingest(db, _root, u.id, size=10)
    _claim(db, job, asset)
    _converts_to(monkeypatch, size=5)

    pipeline._convert_upload(job.id, asset.id)

    db.expire_all()
    assert db.get(DownloadJob, job.id).status == "done"
    assert db.get(MediaAsset, asset.id).size_bytes == 5


# --- the converted file's name and its leftovers -----------------------------------------------


def test_two_dotted_bare_names_keep_a_file_each(db, make_user, _root, monkeypatch):
    # 'Holiday.2024.ä': its suffix is no extension (nothing shell-safe is left of it), so the stem's
    # own dot once read as one — and the converted name lost `_a{id}`, the one thing keeping two apart.
    _probe(monkeypatch)
    u = make_user()
    _converts_to(monkeypatch, size=5)
    finals = []
    for _ in range(2):
        job, asset = _ingest(db, _root, u.id, name="Holiday.2024.ä")
        _claim(db, job, asset)
        pipeline._convert_upload(job.id, asset.id)
        db.expire_all()
        asset = db.get(MediaAsset, asset.id)
        assert asset.status == "ready"
        assert asset.rel_path.endswith(f"_a{asset.id}.mp4"), asset.rel_path
        finals.append(asset.rel_path)
    assert finals[0] != finals[1]
    assert all(storage.abs_path(str(_root), r).exists() for r in finals)


def test_a_bare_names_stem_holds_no_dot_so_a_swapped_suffix_keeps_its_id():
    # Each of the two name rules is held on its own: a new name's stem has no dot, so a sibling
    # derived by `with_suffix` (the poster, the converted file) keeps `_a{id}`.
    rel = storage.upload_rel_path(1, 5, "Holiday.2024.ä")
    assert rel == ".uploads/1/Holiday_2024_a5"
    assert Path(rel).with_suffix(".jpg").name == "Holiday_2024_a5.jpg"


def test_an_upload_placed_with_a_dotted_bare_name_keeps_its_id_when_converted(
    db, make_user, _root, monkeypatch
):
    # A row from before the stems lost their dots: its stem's tail is no extension.
    _probe(monkeypatch)
    u = make_user()
    _converts_to(monkeypatch, size=5)
    job, asset = _ingest(db, _root, u.id, name="Holiday.2024.ä")
    legacy = f".uploads/{u.id}/Holiday.2024_a{asset.id}"
    storage.abs_path(str(_root), asset.rel_path).rename(storage.abs_path(str(_root), legacy))
    asset.rel_path, asset.container = legacy, None
    _claim(db, job, asset)

    pipeline._convert_upload(job.id, asset.id)

    db.expire_all()
    assert db.get(MediaAsset, asset.id).rel_path == f"{legacy}.mp4"


def test_a_conversion_failing_after_it_placed_its_file_removes_it(
    db, make_user, _root, monkeypatch
):
    _probe(monkeypatch)
    u = make_user()
    job, asset = _ingest(db, _root, u.id)
    original = storage.abs_path(str(_root), asset.rel_path)
    _claim(db, job, asset)
    _converts_to(monkeypatch, size=5)

    def _probe_fails(_path):
        raise RuntimeError("probe crashed")

    monkeypatch.setattr(pipeline.probe, "probe_media", _probe_fails)
    pipeline._convert_upload(job.id, asset.id)

    db.expire_all()
    assert db.get(DownloadJob, job.id).status == "error"
    assert original.exists()  # Retry starts from it
    # The row still names the original: a converted file left beside it is one nothing would find.
    assert [p.name for p in _uploads_dir(_root, u.id).iterdir()] == [original.name]


def _finishing_commit_raises(monkeypatch, *, lands: bool) -> None:
    """The commit that makes the converted upload ready raises — after it reached the server
    (`lands`: the connection dropped before its answer) or before (it rolled back)."""
    real_finalize, real_commit = pipeline.service.finalize_if_running, Session.commit
    state = {"session": None, "fired": False}

    def _finalize(db, *a, **k):
        if not state["fired"] and state["session"] is None:
            state["session"] = db
        return real_finalize(db, *a, **k)

    def _commit(self):
        if self is state["session"] and not state["fired"]:
            state["fired"] = True
            if lands:
                real_commit(self)
            raise OperationalError("COMMIT", {}, Exception("connection lost"))
        return real_commit(self)

    monkeypatch.setattr(pipeline.service, "finalize_if_running", _finalize)
    monkeypatch.setattr(Session, "commit", _commit)


def test_a_finishing_commit_that_landed_but_raised_keeps_the_converted_file(
    db, make_user, _root, monkeypatch
):
    # The row names the converted file now: removing it as "never committed" would leave the asset
    # pointing at nothing and its original named by no row, for the sweep to take — both copies gone.
    _probe(monkeypatch)
    u = make_user()
    job, asset = _ingest(db, _root, u.id)
    _claim(db, job, asset)
    _converts_to(monkeypatch, size=5)
    _finishing_commit_raises(monkeypatch, lands=True)

    pipeline._convert_upload(job.id, asset.id)

    db.expire_all()
    asset = db.get(MediaAsset, asset.id)
    assert (db.get(DownloadJob, job.id).status, asset.status) == ("done", "ready")
    assert storage.abs_path(str(_root), asset.rel_path).exists()


def test_a_finishing_commit_that_rolled_back_fails_and_keeps_the_original(
    db, make_user, _root, monkeypatch
):
    _probe(monkeypatch)
    u = make_user()
    job, asset = _ingest(db, _root, u.id)
    original = storage.abs_path(str(_root), asset.rel_path)
    _claim(db, job, asset)
    _converts_to(monkeypatch, size=5)
    _finishing_commit_raises(monkeypatch, lands=False)

    pipeline._convert_upload(job.id, asset.id)

    db.expire_all()
    assert db.get(DownloadJob, job.id).status == "error"
    assert [p.name for p in _uploads_dir(_root, u.id).iterdir()] == [original.name]


def test_an_original_that_cannot_be_removed_leaves_the_upload_ready(
    db, make_user, _root, monkeypatch
):
    # The converted file is committed and the job done: the original's unlink failing must not flip
    # a working asset to an error no Retry is offered for. The sweep takes the original later.
    _probe(monkeypatch)
    u = make_user()
    job, asset = _ingest(db, _root, u.id)
    original = storage.abs_path(str(_root), asset.rel_path)
    _claim(db, job, asset)
    _converts_to(monkeypatch, size=5)
    real_unlink = Path.unlink

    def _unlink(self, missing_ok=False):
        if self == original:
            raise PermissionError("read-only")
        return real_unlink(self, missing_ok=missing_ok)

    monkeypatch.setattr(Path, "unlink", _unlink)
    pipeline._convert_upload(job.id, asset.id)

    db.expire_all()
    asset = db.get(MediaAsset, asset.id)
    assert (db.get(DownloadJob, job.id).status, asset.status) == ("done", "ready")
    assert original.exists()


# --- what the owner is told ------------------------------------------------------------------


def test_a_failed_encode_is_a_failed_conversion_not_a_download_step(
    db, make_user, _root, monkeypatch
):
    _probe(monkeypatch)
    job, asset = _ingest(db, _root, make_user().id)
    _claim(db, job, asset)

    def _ffmpeg_fails(*_a, **_k):
        raise StatedFailure("ffmpeg failed (1): unsupported")

    monkeypatch.setattr(pipeline.editmod, "run_ffmpeg", _ffmpeg_fails)
    pipeline._convert_upload(job.id, asset.id)
    db.expire_all()
    assert db.get(DownloadJob, job.id).error_code == "convert_failed"


def test_a_missing_original_says_so(db, make_user, _root, monkeypatch):
    _probe(monkeypatch)
    job, asset = _ingest(db, _root, make_user().id)
    _claim(db, job, asset)
    storage.abs_path(str(_root), asset.rel_path).unlink()

    pipeline._convert_upload(job.id, asset.id)
    db.expire_all()
    assert db.get(DownloadJob, job.id).error_code == "upload_missing"


def test_the_conversion_reports_its_own_phase(db, make_user, _root, monkeypatch):
    _probe(monkeypatch)
    job, asset = _ingest(db, _root, make_user().id)
    _claim(db, job, asset)
    phases = []
    _converts_to(monkeypatch, size=5, phases=phases)
    pipeline._convert_upload(job.id, asset.id)
    assert phases == ["converting"]  # not yt-dlp's "processing": the queue row shows ffmpeg's %


def test_the_owner_is_told_their_upload_is_ready(db, make_user, _root, monkeypatch):
    _probe(monkeypatch)
    u = make_user()
    job, asset = _ingest(db, _root, u.id)
    _claim(db, job, asset)
    _converts_to(monkeypatch, size=5)
    pipeline._convert_upload(job.id, asset.id)
    notes = db.execute(select(Notification).where(Notification.user_id == u.id)).scalars().all()
    assert [n.data.get("kind") for n in notes] == ["upload"]


def test_an_extension_is_judged_as_it_is_stored(db, make_user, _root, monkeypatch):
    # 'clip.mp4~' is stored as .mp4; judged by the raw 'mp4~' it queued a conversion of a file that
    # was phone-safe already.
    _probe(monkeypatch, vcodec="h264")
    job, asset = _ingest(db, _root, make_user().id, name="clip.mp4~")
    assert (job.status, asset.status, asset.container) == ("done", "ready", "mp4")


# --- the garbage collector ---------------------------------------------------------------------


def _aged(path: Path, hours: float = 5) -> None:
    old = time.time() - hours * 3600
    os.utime(path, (old, old))


def test_the_sweep_takes_a_stranded_upload_file_and_keeps_a_waiting_original(
    db, make_user, _root, monkeypatch
):
    _probe(monkeypatch)
    u = make_user()
    _job, asset = _ingest(db, _root, u.id)
    original = storage.abs_path(str(_root), asset.rel_path)
    stranded = original.with_name("left_by_a_stopped_conversion_a999.mp4")
    stranded.write_bytes(b"x")
    _aged(original)
    _aged(stranded)

    gcmod._sweep_orphan_files(db, str(_root))

    assert original.exists()  # queued for conversion: its row names it, ready or not
    assert not stranded.exists()  # no row names it


@pytest.mark.parametrize(
    ("job_status", "asset_status"),
    [
        ("queued", "pending"),  # waiting for the worker
        ("running", "downloading"),  # being converted, for longer than the sweep's age floor
        ("error", "error"),  # failed, kept for Retry
        ("done", "ready"),  # converted
    ],
)
def test_the_sweep_keeps_an_aged_upload_file_its_row_names_in_any_state(
    job_status, asset_status, db, make_user, _root, monkeypatch
):
    _probe(monkeypatch)
    job, asset = _ingest(db, _root, make_user().id)
    path = storage.abs_path(str(_root), asset.rel_path)
    job.status, asset.status = job_status, asset_status
    db.commit()
    _aged(path)

    gcmod._sweep_orphan_files(db, str(_root))

    assert path.exists()


def _row(db, asset_id: int) -> MediaAsset | None:
    """The asset as the database has it now (a `get` would refresh the session's own copy, and
    raise for one deleted under it)."""
    db.expire_all()
    return db.execute(select(MediaAsset).where(MediaAsset.id == asset_id)).scalar_one_or_none()


def _route_moved(db, job, asset, job_status: str, ref_count: int) -> None:
    """What the cancel/pause/delete route leaves behind, written as SQL: the session's copy of
    `ref_count` is stale (the routes count it in SQL), so an ORM assignment of the same value would
    write nothing."""
    db.execute(update(DownloadJob).where(DownloadJob.id == job.id).values(status=job_status))
    db.execute(update(MediaAsset).where(MediaAsset.id == asset.id).values(ref_count=ref_count))
    db.commit()


def _stopped_by(monkeypatch, db, job, asset, job_status: str, ref_count: int) -> None:
    """The conversion is stopped mid-encode after the route moved the job to `job_status` and its
    hold on the asset to `ref_count` (cancel and delete release it; pause keeps it)."""

    def _encode(*_a, **_k):
        _route_moved(db, job, asset, job_status, ref_count)
        raise pipeline.editmod.EditAborted()

    monkeypatch.setattr(pipeline, "_ensure_phone_safe", _encode)


def test_an_upload_cancelled_mid_conversion_is_freed_once_the_conversion_stops(
    db, make_user, _root, monkeypatch
):
    # `release_asset` leaves it while the conversion reads the file; once it stopped, nothing holds it
    # — kept, it would sit on disk outside every footprint until the GC's upload pass.
    _probe(monkeypatch)
    u = make_user()
    job, asset = _ingest(db, _root, u.id, size=10)
    original = storage.abs_path(str(_root), asset.rel_path)
    _claim(db, job, asset)
    _stopped_by(monkeypatch, db, job, asset, "canceled", 0)

    pipeline._convert_upload(job.id, asset.id)

    assert _row(db, asset.id) is None
    assert not original.exists()
    assert quota.footprint(db, u.id) == 0


def test_an_upload_paused_mid_conversion_keeps_its_original_for_the_resume(
    db, make_user, _root, monkeypatch
):
    _probe(monkeypatch)
    u = make_user()
    job, asset = _ingest(db, _root, u.id, size=10)
    original = storage.abs_path(str(_root), asset.rel_path)
    _claim(db, job, asset)
    _stopped_by(monkeypatch, db, job, asset, "paused", 1)

    pipeline._convert_upload(job.id, asset.id)

    db.expire_all()
    assert db.get(MediaAsset, asset.id).status == "pending"
    assert original.exists()


def test_an_upload_whose_conversion_finished_after_its_job_was_cancelled_is_freed(
    db, make_user, _root, monkeypatch
):
    _probe(monkeypatch)
    u = make_user()
    job, asset = _ingest(db, _root, u.id, size=10)
    _claim(db, job, asset)
    _converts_to(monkeypatch, size=5)
    real_encode = pipeline._ensure_phone_safe

    def _encode_then_cancelled(*a, **k):
        out = real_encode(*a, **k)
        _route_moved(db, job, asset, "canceled", 0)
        return out

    monkeypatch.setattr(pipeline, "_ensure_phone_safe", _encode_then_cancelled)
    pipeline._convert_upload(job.id, asset.id)

    assert _row(db, asset.id) is None
    left = _uploads_dir(_root, u.id)
    assert not left.exists() or list(left.iterdir()) == []


def test_the_cache_cap_never_evicts_an_upload(db, make_user, _root, monkeypatch):
    _probe(monkeypatch, vcodec="h264")
    u = make_user()
    _job, upload = _ingest(db, _root, u.id, name="mine.mp4", size=10)
    db.add(
        MediaAsset(
            source_kind="youtube", source_ref="vid1", format_sig="best", status="ready",
            rel_path="Chan/clip.mp4", size_bytes=10, title="downloaded",
            last_access_at=datetime.now(timezone.utc),
        )
    )  # fmt: skip
    # The upload is the least recently used: an LRU pass that may take it takes it first.
    upload.last_access_at = datetime.now(timezone.utc) - timedelta(days=30)
    db.commit()
    (_root / "Chan").mkdir()
    (_root / "Chan" / "clip.mp4").write_bytes(b"d" * 10)
    sysconfig.set_value(db, "download_total_max_bytes", 15)  # 20 on disk
    db.commit()

    gcmod.run_download_gc(db)

    db.expire_all()
    assert db.get(MediaAsset, upload.id).status == "ready"
    assert storage.abs_path(str(_root), upload.rel_path).exists()


# --- a share link ----------------------------------------------------------------------------


@pytest.mark.parametrize(
    ("job_status", "asset_status", "reason"),
    [
        ("queued", "pending", "video_preparing"),  # a re-download waiting for the worker
        ("running", "downloading", "video_preparing"),  # being made now
        ("canceled", "pending", "video_gone"),  # a rebuild cancelled: nothing will finish it
        ("paused", "pending", "video_gone"),  # held until its owner resumes it, maybe never
    ],
)
def test_a_link_says_preparing_only_while_its_video_is_being_made(
    job_status, asset_status, reason, db, as_user, make_user, _root, monkeypatch, upload_headers
):
    # A link is made on a finished job; its asset can be (re)made under it later — a shared cache row
    # re-fetched, a conversion retried. While that runs the page says "try again shortly"; once
    # nothing is making it, it says the video is gone rather than promising it forever.
    _probe(monkeypatch, vcodec="h264")
    c = as_user(make_user())
    job = c.post(
        "/api/files/upload", content=b"h264 bytes", headers=upload_headers("clip.mp4")
    ).json()
    link = c.post(f"/api/downloads/{job['id']}/links", json={"allow_download": False}).json()
    assert c.get(f"/api/public/watch/{link['token']}").status_code == 200
    db.get(MediaAsset, job["asset"]["id"]).status = asset_status
    db.get(DownloadJob, job["id"]).status = job_status
    db.commit()
    r = c.get(f"/api/public/watch/{link['token']}")
    assert r.status_code == 404
    assert r.json()["detail"]["reason"] == reason


def test_a_link_to_a_suspended_owners_waiting_rebuild_says_it_is_gone(
    db, as_user, make_user, _root, monkeypatch, upload_headers
):
    # The worker leaves a suspended user's queued jobs where they are, maybe for good: like a paused
    # rebuild, nothing is making the video, so "try again shortly" would never come true.
    _probe(monkeypatch, vcodec="h264")
    owner = make_user()
    c = as_user(owner)
    job = c.post(
        "/api/files/upload", content=b"h264 bytes", headers=upload_headers("clip.mp4")
    ).json()
    link = c.post(f"/api/downloads/{job['id']}/links", json={"allow_download": False}).json()
    db.get(MediaAsset, job["asset"]["id"]).status = "pending"
    db.get(DownloadJob, job["id"]).status = "queued"
    owner.is_suspended = True
    db.commit()
    r = c.get(f"/api/public/watch/{link['token']}")
    assert r.status_code == 404
    assert r.json()["detail"]["reason"] == "video_gone"
    # A rebuild already running when the owner was suspended still finishes.
    db.get(DownloadJob, job["id"]).status = "running"
    db.commit()
    assert c.get(f"/api/public/watch/{link['token']}").json()["detail"]["reason"] == (
        "video_preparing"
    )


def test_the_admin_storage_view_counts_a_waiting_upload(db, as_user, make_user, _root, monkeypatch):
    # The same disk the owner's footprint and the GC's cap count (`quota.ON_DISK`).
    _probe(monkeypatch)
    u = make_user()
    _ingest(db, _root, u.id, size=10)
    admin = make_user(role="admin")
    r = as_user(admin).get("/api/admin/downloads/storage").json()
    assert (r["files_on_disk"], r["total_bytes"]) == (1, 10)
    assert {p["user_id"]: p["footprint_bytes"] for p in r["per_user"]} == {u.id: 10, admin.id: 0}


def test_the_admin_storage_view_names_each_users_quota_and_files(
    db, as_user, make_user, _root, monkeypatch
):
    # A row says how full the user is against THEIR limit (a custom one, here), and how many files
    # their Files lists — the finished ones, not one still waiting for its conversion.
    _probe(monkeypatch)
    u = make_user()
    done, _ = _ingest(db, _root, u.id, size=10)
    waiting, _ = _ingest(db, _root, u.id, name="b.mov", size=5)
    done.status, waiting.status = "done", "queued"
    db.commit()
    admin = as_user(make_user(role="admin"))
    put = admin.put(f"/api/admin/downloads/quota/{u.id}", json={"max_bytes": 1000})
    assert put.status_code == 200
    rows = admin.get("/api/admin/downloads/storage").json()["per_user"]
    mine = [(p["files"], p["max_bytes"], p["unlimited"]) for p in rows if p["user_id"] == u.id]
    assert mine == [(1, 1000, False)]
    # Every download names its owner, so its row can open THEIR quota.
    jobs = admin.get("/api/admin/downloads").json()
    assert {(j["id"], j["user_id"]) for j in jobs} == {(done.id, u.id), (waiting.id, u.id)}


def test_the_admin_storage_view_lists_every_user_even_one_without_a_file(db, as_user, make_user):
    # The admin sets a quota BEFORE anyone downloads: a user holding nothing is still a row.
    empty = make_user()
    admin_user = make_user(role="admin")
    admin = as_user(admin_user)
    rows = admin.get("/api/admin/downloads/storage").json()["per_user"]
    by_id = {p["user_id"]: p for p in rows}
    assert {empty.id, admin_user.id} <= by_id.keys()
    assert (by_id[empty.id]["email"], by_id[empty.id]["files"]) == (empty.email, 0)
    assert by_id[empty.id]["footprint_bytes"] == 0


@pytest.mark.parametrize(
    ("job_status", "asset_status", "ref_count"),
    [
        ("done", "ready", 1),  # a finished upload its owner keeps
        ("queued", "pending", 1),  # one waiting for the worker
        ("canceled", "downloading", 0),  # cancelled mid-conversion: the worker still reads it
    ],
)
def test_the_upload_reap_leaves_one_still_held_or_still_read(
    job_status, asset_status, ref_count, db, make_user, _root, monkeypatch
):
    _probe(monkeypatch)
    job, asset = _ingest(db, _root, make_user().id)
    path = storage.abs_path(str(_root), asset.rel_path)
    asset.status = asset_status
    db.commit()
    _route_moved(db, job, asset, job_status, ref_count)

    gcmod.run_download_gc(db)

    assert _row(db, asset.id) is not None
    assert path.exists()