siftlode/backend/tests/test_download_limits.py

"""DB-backed tests for R12 · S5 backend residue — the two download limits:

- **#2 claim fairness** (`worker._claim_job`): users already AT their per-user concurrency limit
  are excluded IN THE SQL, so one user flooding the queue past the fetch window can't starve
  everyone behind them. The old fixed `LIMIT 50` window did exactly that.
- **#3 list cap** (`routes.downloads.list_downloads` / `shared_with_me`): the per-user listings
  are bounded so a heavy library can't return an unbounded result set.

Uses the `db` fixture (real Postgres) — skipped when no database is reachable. `_claim_job` runs in
its OWN session, so every test COMMITS its setup before calling it.
"""

from app.downloads import quota
from app.models import DownloadJob, DownloadQuota, MediaAsset, User
from app.worker import _claim_job


def _job(db, user_id, queue_pos, status="queued", asset_id=None):
    j = DownloadJob(
        user_id=user_id,
        asset_id=asset_id,
        source_kind="youtube",
        source_ref="x" * 11,
        status=status,
        queue_pos=queue_pos,
        profile_snapshot={},
    )
    db.add(j)
    return j


def _asset(db, status, ref):
    a = MediaAsset(source_kind="url", source_ref=ref, format_sig="s", status=status, ref_count=1)
    db.add(a)
    db.flush()
    return a


def _quota(db, user_id, max_concurrent, unlimited=False):
    db.add(
        DownloadQuota(
            user_id=user_id,
            max_bytes=10**12,
            max_concurrent=max_concurrent,
            max_jobs=10**6,
            unlimited=unlimited,
        )
    )


# --- #2 · claim fairness --------------------------------------------------------------------


def test_capped_user_flood_does_not_starve_users_behind_the_window(db):
    a = User(email="flooder@example.com")
    b = User(email="waiter@example.com")
    db.add_all([a, b])
    db.flush()
    _quota(
        db, a.id, max_concurrent=1
    )  # A: 1 concurrent, already running one -> all A's queued are ineligible
    _quota(db, b.id, max_concurrent=3)
    _job(db, a.id, queue_pos=0, status="running")
    for i in range(1, 61):  # A floods the FRONT with far more than the 50-row fetch window
        _job(db, a.id, queue_pos=i)
    bjob = _job(db, b.id, queue_pos=100)  # B sits behind the flood
    db.commit()

    assert _claim_job() == bjob.id  # B is served despite A's 60-job flood filling the old window
    db.expire_all()
    assert db.get(DownloadJob, bjob.id).status == "running"


def test_eligible_user_claims_oldest_by_queue_pos(db):
    u = User(email="solo@example.com")
    db.add(u)
    db.flush()
    _quota(db, u.id, max_concurrent=2)
    first = _job(db, u.id, queue_pos=1)
    _job(db, u.id, queue_pos=2)
    db.commit()
    assert _claim_job() == first.id


def test_all_candidates_capped_returns_none(db):
    u = User(email="capped@example.com")
    db.add(u)
    db.flush()
    _quota(db, u.id, max_concurrent=1)
    _job(db, u.id, queue_pos=0, status="running")  # at the cap
    _job(db, u.id, queue_pos=1)  # queued but ineligible
    db.commit()
    assert _claim_job() is None


def test_stored_zero_concurrency_is_floored_not_starving(db):
    # A stray max_concurrent=0 is only reachable by a direct DB edit (admin_set_quota floors at 1),
    # but it must not wedge the claim loop. The effective limit floors to 1 on BOTH the read path
    # (resolve -> at_concurrency_limit) and in the claim SQL (GREATEST(..., 1)), so the two "at
    # limit" checks agree: the user's own first job is claimable and its flood never monopolizes
    # the window with permanently-ineligible jobs. (Before the floor, _claim_job returned None here
    # — the guard rejected all 60 of A's jobs — and any user behind them starved.)
    a = User(email="zero@example.com")
    b = User(email="behind@example.com")
    db.add_all([a, b])
    db.flush()
    _quota(db, a.id, max_concurrent=0)  # nonsensical stored value
    _quota(db, b.id, max_concurrent=2)
    for i in range(60):
        _job(db, a.id, queue_pos=i)
    _job(db, b.id, queue_pos=100)
    db.commit()

    claimed = _claim_job()
    assert claimed is not None  # the loop is not wedged
    db.expire_all()
    assert db.get(DownloadJob, claimed).user_id == a.id  # A's own first job runs (floored to 1)


def test_job_whose_asset_is_already_downloading_is_skipped(db):
    """B-sub-3: a queued job whose shared asset is ALREADY `downloading` (another worker owns it) must
    not be claimed — otherwise a second worker claims it, finds the asset busy, requeues, and ping-
    pongs on it, starving everyone behind. The claim SQL excludes it, so the job behind it (on a
    different, free asset) is served instead."""
    u = User(email="asset-excl@example.com")
    db.add(u)
    db.flush()
    _quota(
        db, u.id, max_concurrent=5
    )  # high → the ASSET rule, not the per-user rule, does the excluding
    busy = _asset(db, status="downloading", ref="busy")
    free = _asset(db, status="pending", ref="free")
    _job(
        db, u.id, queue_pos=1, status="running", asset_id=busy.id
    )  # the holder actually downloading it
    competitor = _job(db, u.id, queue_pos=2, asset_id=busy.id)  # same asset — must be skipped
    behind = _job(db, u.id, queue_pos=3, asset_id=free.id)  # different asset — should be served
    db.commit()

    assert _claim_job() == behind.id  # the competitor on the busy asset was skipped
    db.expire_all()
    assert db.get(DownloadJob, competitor.id).status == "queued"  # still queued, never claimed


def test_unlimited_user_is_never_excluded(db):
    u = User(email="unlimited@example.com")
    db.add(u)
    db.flush()
    _quota(db, u.id, max_concurrent=1, unlimited=True)
    for i in range(3):  # 3 running, well past max_concurrent — but unlimited bypasses the cap
        _job(db, u.id, queue_pos=i, status="running")
    q = _job(db, u.id, queue_pos=10)
    db.commit()
    assert _claim_job() == q.id


# --- #3 · list cap --------------------------------------------------------------------------


def test_list_downloads_is_capped(db, monkeypatch):
    from app.routes import downloads as dl

    u = User(email="heavy@example.com")
    db.add(u)
    db.flush()
    for i in range(3):
        _job(db, u.id, queue_pos=i, status="done")
    db.commit()

    monkeypatch.setattr(dl, "_MAX_LIST", 2)
    rows = dl.list_downloads(user=u, db=db)
    assert len(rows) == 2  # bounded by the cap, not the 3 rows on disk


# --- E12 · S3 · the worker-side max_bytes remainder ------------------------------------------


def _sized_asset(db, size_bytes, ref):
    a = MediaAsset(
        source_kind="url",
        source_ref=ref,
        format_sig="s",
        status="ready",
        ref_count=1,
        size_bytes=size_bytes,
    )
    db.add(a)
    db.flush()
    return a


def test_remaining_bytes_is_none_for_an_unlimited_user(db):
    u = User(email="unlimited@example.com")
    db.add(u)
    db.flush()
    _quota(db, u.id, max_concurrent=2, unlimited=True)
    db.flush()
    # None, not a number: "no cap" and "no budget left" must never be the same value downstream —
    # build_ydl_opts leaves max_filesize unset for both, so only the caller can tell them apart.
    assert quota.remaining_bytes(db, u.id) is None


def test_remaining_bytes_is_the_unspent_part_of_the_cap(db):
    u = User(email="partial@example.com")
    db.add(u)
    db.flush()
    db.add(
        DownloadQuota(user_id=u.id, max_bytes=1000, max_concurrent=2, max_jobs=100, unlimited=False)
    )
    a = _sized_asset(db, 400, "partial-1")
    _job(db, u.id, queue_pos=0, status="done", asset_id=a.id)
    db.flush()
    assert quota.remaining_bytes(db, u.id) == 600


def test_remaining_bytes_is_zero_once_the_cap_is_met(db):
    # THE case this exists for: `check_enqueue` let these jobs in while the user was under the cap,
    # and an earlier one has filled it since. Zero is what makes the worker refuse; if this ever
    # returned the raw negative or None instead, the user already over quota would be the only one
    # downloading with no size ceiling at all.
    u = User(email="atcap@example.com")
    db.add(u)
    db.flush()
    db.add(
        DownloadQuota(user_id=u.id, max_bytes=1000, max_concurrent=2, max_jobs=100, unlimited=False)
    )
    a = _sized_asset(db, 1500, "atcap-1")  # overshot: one job finished larger than the remainder
    _job(db, u.id, queue_pos=0, status="done", asset_id=a.id)
    db.flush()
    assert quota.remaining_bytes(db, u.id) == 0