peter
b04b51c909
A related list's cursor cost nothing, and a cursor is a free string: a replayed one read and enriched without end. Every request now takes a slot of the user's daily cap — taken under a lock before the work (`quota.claim_action`), so requests sent together cannot all pass the last free one, and given back when the request fails outright (`quota.refund_action`). The live search's scrape source takes its slots the same way. A limit of 0 turns the feature off: a distinct 409 `related_disabled` / `search_disabled`, and `/api/me` says `related_enabled` / `search_enabled` so the UI can leave it out. A videos.list that fails no longer reads as an empty page: `ingest_candidates` raises, and the search and the related list answer the refusal (the shared budget spent says "quota") instead of "no results". The two routes' page loops are one (`_gather`). `clear` deletes only what the caller found: a related list's rows carry no find, and any user could name them.
133 lines
4.5 KiB
Python
133 lines
4.5 KiB
Python
"""`GET /api/search/youtube` — the per-user daily cap of the live search (scrape source).
|
|
|
|
The scrape is replaced by pages the test hands out, and YouTube's videos.list by an offline enrich.
|
|
What is held: every request takes a slot, a cursor's too; a search that failed outright gives its
|
|
slot back; a zero limit turns live search off, which says so instead of "try again tomorrow"; and a
|
|
lookup that fails says why instead of answering "no results"."""
|
|
|
|
import pytest
|
|
|
|
from app import quota, sysconfig
|
|
from app.routes import search as search_routes
|
|
from app.youtube.client import YouTubeError
|
|
from app.youtube.innertube import ScrapeError
|
|
|
|
ACTION = quota.QuotaAction.VIDEOS_SEARCH
|
|
|
|
|
|
def _item(n: int) -> dict:
|
|
return {
|
|
"id": f"srch{n:07d}",
|
|
"channel_id": "UCsearch001",
|
|
"channel_title": "Chan",
|
|
"title": f"Found {n}",
|
|
"published_at": None,
|
|
"thumbnail_url": None,
|
|
"live_broadcast": "none",
|
|
}
|
|
|
|
|
|
class _FakeYT:
|
|
fail_videos = False
|
|
|
|
def __init__(self, *_a, **_kw):
|
|
pass
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
return False
|
|
|
|
def get_channels(self, ids):
|
|
return []
|
|
|
|
def get_videos(self, ids):
|
|
if _FakeYT.fail_videos:
|
|
raise YouTubeError("videos -> 403", reason="quotaExceeded", path="videos")
|
|
return [
|
|
{
|
|
"id": i,
|
|
"snippet": {"title": f"T {i}", "publishedAt": "2025-01-01T00:00:00Z"},
|
|
"contentDetails": {"duration": "PT30M"},
|
|
"statistics": {},
|
|
"status": {},
|
|
}
|
|
for i in ids
|
|
]
|
|
|
|
|
|
@pytest.fixture
|
|
def pages(monkeypatch, db):
|
|
"""The scrape answers with the pages in `state["pages"]`, one per call."""
|
|
sysconfig.set_value(db, "shorts_probe_max_seconds", 60)
|
|
sysconfig.set_value(db, "search_source", "scrape")
|
|
state: dict = {"pages": [], "calls": 0}
|
|
|
|
def search(term, continuation=None):
|
|
state["calls"] += 1
|
|
page = state["pages"].pop(0)
|
|
if isinstance(page, Exception):
|
|
raise page
|
|
return page
|
|
|
|
monkeypatch.setattr(search_routes.search_scrape, "search", search)
|
|
monkeypatch.setattr(search_routes, "YouTubeClient", _FakeYT)
|
|
monkeypatch.setattr(_FakeYT, "fail_videos", False)
|
|
return state
|
|
|
|
|
|
def _page(n: int) -> dict:
|
|
return {"items": [_item(n)], "next_page_token": None}
|
|
|
|
|
|
def test_every_request_takes_a_slot_a_cursors_too(db, make_user, as_user, pages):
|
|
user = make_user()
|
|
sysconfig.set_value(db, "search_daily_limit_per_user", 1)
|
|
pages["pages"] = [_page(1), _page(2)]
|
|
client = as_user(user)
|
|
assert client.get("/api/search/youtube?q=cats").status_code == 200
|
|
refused = client.get("/api/search/youtube?q=cats&cursor=more")
|
|
assert refused.status_code == 429
|
|
assert refused.json()["detail"]["reason"] == "search_daily_limit"
|
|
assert pages["calls"] == 1
|
|
|
|
|
|
def test_a_search_that_failed_gives_its_slot_back(db, make_user, as_user, pages):
|
|
user = make_user()
|
|
sysconfig.set_value(db, "search_daily_limit_per_user", 1)
|
|
pages["pages"] = [ScrapeError("blocked"), _page(1)]
|
|
client = as_user(user)
|
|
assert client.get("/api/search/youtube?q=cats").status_code == 422
|
|
assert quota.actions_today(db, user.id, ACTION) == 0
|
|
assert client.get("/api/search/youtube?q=cats").status_code == 200
|
|
assert quota.actions_today(db, user.id, ACTION) == 1
|
|
|
|
|
|
def test_a_zero_limit_turns_live_search_off(db, make_user, as_user, pages):
|
|
user = make_user()
|
|
sysconfig.set_value(db, "search_daily_limit_per_user", 0)
|
|
resp = as_user(user).get("/api/search/youtube?q=cats")
|
|
assert resp.status_code == 409
|
|
assert resp.json()["detail"] == {"code": "refused", "reason": "search_disabled"}
|
|
assert pages["calls"] == 0
|
|
|
|
|
|
def test_a_zero_limit_turns_the_api_source_off_too(db, make_user, as_user, pages):
|
|
user = make_user()
|
|
sysconfig.set_value(db, "search_source", "api")
|
|
sysconfig.set_value(db, "search_daily_limit_per_user", 0)
|
|
resp = as_user(user).get("/api/search/youtube?q=cats")
|
|
assert resp.status_code == 409
|
|
assert resp.json()["detail"]["reason"] == "search_disabled"
|
|
|
|
|
|
def test_a_lookup_that_fails_says_why_not_no_results(db, make_user, as_user, pages, monkeypatch):
|
|
user = make_user()
|
|
monkeypatch.setattr(_FakeYT, "fail_videos", True)
|
|
pages["pages"] = [_page(1)]
|
|
resp = as_user(user).get("/api/search/youtube?q=cats")
|
|
assert resp.status_code == 422
|
|
assert resp.json()["detail"] == {"code": "youtube", "reason": "quota"}
|
|
assert quota.actions_today(db, user.id, ACTION) == 0
|