siftlode/backend/tests/test_abuse_limits.py

"""The route-level halves of SEC-12 and SEC-14.

`test_urlguard.py` pins the address judgement itself; this pins that the ENQUEUE route actually
consults it and answers in the structured `{code, reason}` shape the frontend localizes, and that a
manual sync trigger stops spending the instance-wide YouTube budget when held down.

DB-backed (the `as_user` fixture needs real rows) — skipped when no database is reachable.
"""

from app.downloads import urlguard


def test_enqueueing_a_loopback_url_is_refused_with_a_structured_detail(
    db, as_user, make_user, monkeypatch
):
    user = make_user()
    db.commit()
    monkeypatch.setattr(
        urlguard.socket, "getaddrinfo", lambda *a, **kw: [(None, None, None, "", ("127.0.0.1", 0))]
    )
    r = as_user(user).post("/api/downloads", json={"source": "http://redis.internal:6379/x"})
    assert r.status_code == 422
    # The shape matters as much as the refusal: a bare English string under a bilingual UI is the
    # thing the canonical-patterns table forbids.
    assert r.json()["detail"] == {"code": "url_blocked", "reason": urlguard.REASON_PRIVATE}


def test_a_public_url_gets_past_the_guard(db, as_user, make_user, monkeypatch):
    # Mutation check for the test above: with a public answer the request must fail LATER (or
    # succeed), never with url_blocked — otherwise the guard would be refusing everything and the
    # first test would still pass.
    user = make_user()
    db.commit()
    monkeypatch.setattr(
        urlguard.socket,
        "getaddrinfo",
        lambda *a, **kw: [(None, None, None, "", ("93.184.216.34", 0))],
    )
    # An inline spec, so nothing ELSE refuses it: without one the request stopped at the
    # profile lookup (400 no_profile) and "not url_blocked" held for a request the guard never
    # had to wave through.
    r = as_user(user).post(
        "/api/downloads", json={"source": "https://example.com/v", "spec": {"format": "best"}}
    )
    assert r.status_code == 200, r.text
    assert r.json()["status"] == "queued"


def test_youtube_sources_never_consult_the_guard(db, as_user, make_user, monkeypatch):
    # A YouTube job resolves to a fixed public host from an 11-char id, so there is no
    # user-controlled destination — resolving one would be pure latency on every enqueue.
    def boom(*a, **kw):
        raise AssertionError("the URL guard resolved DNS for a YouTube source")

    user = make_user()
    db.commit()
    monkeypatch.setattr(urlguard.socket, "getaddrinfo", boom)
    # The job must actually be created: a request refused for any other reason would never reach
    # the guard either, and the test would pass without proving anything.
    r = as_user(user).post(
        "/api/downloads", json={"source": "dQw4w9WgXcQ", "spec": {"format": "best"}}
    )
    assert r.status_code == 200, r.text
    assert r.json()["source_kind"] == "youtube"


def test_manual_subscription_sync_goes_on_cooldown(db, as_user, make_user, monkeypatch):
    from app.routes import sync as syncroutes

    user = make_user()
    db.commit()
    # Neither the YouTube grant nor the import itself is under test — the limiter is.
    monkeypatch.setattr(syncroutes, "has_read_scope", lambda u: True)
    monkeypatch.setattr(syncroutes, "import_subscriptions", lambda d, u: {"imported": 0})

    c = as_user(user)
    codes = [c.post("/api/sync/subscriptions").status_code for _ in range(4)]
    # 3 per 10 minutes: the first three pass, the fourth is refused.
    assert codes[:3] == [200, 200, 200]
    assert codes[3] == 429
    assert c.post("/api/sync/subscriptions").json()["detail"] == {"code": "sync_cooldown"}


def test_the_cooldown_is_per_user_not_global(db, as_user, make_user, monkeypatch):
    # Keyed by user id on purpose: the shared YouTube budget is spent per account, and one user
    # exhausting their allowance must not lock everyone else out of their own sync button.
    from app.routes import sync as syncroutes

    a, b = make_user(), make_user()
    db.commit()
    monkeypatch.setattr(syncroutes, "has_read_scope", lambda u: True)
    monkeypatch.setattr(syncroutes, "import_subscriptions", lambda d, u: {"imported": 0})

    ca = as_user(a)
    assert [ca.post("/api/sync/subscriptions").status_code for _ in range(4)] == [
        200,
        200,
        200,
        429,
    ]
    assert ca.post("/api/sync/subscriptions").status_code == 429
    assert as_user(b).post("/api/sync/subscriptions").status_code == 200


def test_the_single_playlist_repull_shares_the_pull_cooldown(db, as_user, make_user, monkeypatch):
    # SEC-14, the sweep half: `revert-youtube` spends under the SAME QuotaAction.PLAYLISTS_PULL key
    # as the bulk `sync-youtube`, so limiting only one of them left the identical drain open one
    # route over. Both must draw on the one allowance.
    from app.models import Playlist
    from app.routes import playlists as plroutes

    user = make_user()
    pl = Playlist(user_id=user.id, name="Linked", source="youtube", yt_playlist_id="PL123")
    db.add(pl)
    db.commit()
    monkeypatch.setattr(plroutes, "has_read_scope", lambda u: True)
    monkeypatch.setattr(plroutes, "repull_playlist", lambda d, u, p: {"ok": True})
    monkeypatch.setattr(plroutes, "pull_user_playlists", lambda d, u: {"ok": True})

    c = as_user(user)
    codes = [c.post(f"/api/playlists/{pl.id}/revert-youtube").status_code for _ in range(4)]
    assert codes[:3] == [200, 200, 200]
    assert codes[3] == 429
    # One allowance, not one each: the bulk route is now on cooldown too.
    r = c.post("/api/playlists/sync-youtube")
    assert r.status_code == 429
    assert r.json()["detail"] == {"code": "sync_cooldown"}