peter
2e1934689f
The slot was kept as soon as search.list was about to be called, so a search refused before it went out (no connection, no token) cost the user a slot for nothing. It is kept now only once the client recorded the call's units, or the request went out and its answer never came (YouTube may have counted it): YouTubeError says whether a transport failure was sent, and the client counts what it spent. The process lock around the budget check and the call is waited for at most ten seconds; past that the search is refused as busy (search_busy, a retry can help) and gets its slot back, instead of parking a request worker behind every slow call before it. The lock test now pins one hold around both steps, and the busy test fails on a wait without a bound.
362 lines
14 KiB
Python
362 lines
14 KiB
Python
"""`GET /api/search/youtube` — the per-user daily cap of the live search, both sources.
|
|
|
|
The scrape and search.list are 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"; a lookup that fails says why instead of answering "no results"; only a few
|
|
failures an hour get their slot back; and the API source — 100 units a page — reads one page per
|
|
request, is refused before the call when the shared budget cannot pay for it (checked again under
|
|
one lock with the call, whose wait is bounded), counts one slot per search though search.list logs
|
|
its own event, and keeps the slot of a search search.list was already paid for — but not of one
|
|
that never reached YouTube."""
|
|
|
|
import threading
|
|
|
|
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):
|
|
self.spent = 0
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
return False
|
|
|
|
def get_channels(self, ids):
|
|
return []
|
|
|
|
# search.list: the pages the test hands out, each logged as the real client does (100 units,
|
|
# under the action the route attributes it to) — a refusal YouTube answered too, but not a
|
|
# failure no endpoint answered (no connection, no token).
|
|
api_pages: list = []
|
|
api_calls = 0
|
|
|
|
def search_videos(self, q, page_token=None, order="relevance"):
|
|
_FakeYT.api_calls += 1
|
|
page = _FakeYT.api_pages.pop(0)
|
|
if not (isinstance(page, YouTubeError) and not page.path):
|
|
quota.record_usage(100)
|
|
self.spent += 100
|
|
if isinstance(page, Exception):
|
|
raise page
|
|
return page
|
|
|
|
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)
|
|
monkeypatch.setattr(_FakeYT, "api_pages", [])
|
|
monkeypatch.setattr(_FakeYT, "api_calls", 0)
|
|
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
|
|
|
|
|
|
def _api(db) -> None:
|
|
sysconfig.set_value(db, "search_source", "api")
|
|
sysconfig.set_value(db, "quota_daily_budget", 10000)
|
|
|
|
|
|
def test_the_api_source_reads_one_page_per_request(db, make_user, as_user, pages):
|
|
# Each page is 100 units: a short page is answered as it is, with its cursor, never paged on.
|
|
_api(db)
|
|
_FakeYT.api_pages = [{"items": [_item(1)], "next_page_token": "nextpage"}, _page(2)]
|
|
resp = as_user(make_user()).get("/api/search/youtube?q=cats&limit=20")
|
|
assert resp.status_code == 200
|
|
assert _FakeYT.api_calls == 1
|
|
assert resp.json()["next_cursor"] == "nextpage"
|
|
|
|
|
|
def test_the_api_source_counts_one_slot_per_search(db, make_user, as_user, pages):
|
|
# search.list logs its own event under the action (record_usage): accounting, not a second slot.
|
|
user = make_user()
|
|
_api(db)
|
|
sysconfig.set_value(db, "search_daily_limit_per_user", 2)
|
|
_FakeYT.api_pages = [_page(1), _page(2), _page(3)]
|
|
client = as_user(user)
|
|
assert client.get("/api/search/youtube?q=cats").status_code == 200
|
|
assert quota.actions_today(db, user.id, ACTION) == 1
|
|
assert client.get("/api/search/youtube?q=dogs").status_code == 200
|
|
refused = client.get("/api/search/youtube?q=birds")
|
|
assert refused.status_code == 429
|
|
assert refused.json()["detail"]["reason"] == "search_daily_limit"
|
|
assert _FakeYT.api_calls == 2
|
|
|
|
|
|
def test_the_api_source_takes_its_slot_under_the_lock(db, make_user, as_user, pages, monkeypatch):
|
|
user = make_user()
|
|
_api(db)
|
|
claimed: list = []
|
|
real = quota.claim_action
|
|
|
|
def claim(uid, action, limit):
|
|
claimed.append(action)
|
|
return real(uid, action, limit)
|
|
|
|
monkeypatch.setattr(search_routes.quota, "claim_action", claim)
|
|
_FakeYT.api_pages = [_page(1)]
|
|
assert as_user(user).get("/api/search/youtube?q=cats").status_code == 200
|
|
assert claimed == [ACTION]
|
|
|
|
|
|
def test_the_api_source_is_refused_before_the_call_when_the_budget_cannot_pay(
|
|
db, make_user, as_user, pages
|
|
):
|
|
user = make_user()
|
|
_api(db)
|
|
sysconfig.set_value(db, "quota_daily_budget", 50)
|
|
_FakeYT.api_pages = [_page(1)]
|
|
resp = as_user(user).get("/api/search/youtube?q=cats")
|
|
assert resp.status_code == 429
|
|
assert resp.json()["detail"]["reason"] == "quota_spent"
|
|
assert _FakeYT.api_calls == 0
|
|
assert quota.actions_today(db, user.id, ACTION) == 0
|
|
|
|
|
|
def test_an_api_search_that_search_list_refused_keeps_its_slot(db, make_user, as_user, pages):
|
|
# YouTube charges a refused call too (the client records the units first), so a search that
|
|
# reached search.list has spent the shared budget: refunding it let a bogus cursor spend it all.
|
|
user = make_user()
|
|
_api(db)
|
|
sysconfig.set_value(db, "search_daily_limit_per_user", 1)
|
|
_FakeYT.api_pages = [
|
|
YouTubeError("search -> 400", reason="invalidPageToken", path="search"),
|
|
_page(1),
|
|
]
|
|
client = as_user(user)
|
|
assert client.get("/api/search/youtube?q=cats&cursor=bogus").status_code == 422
|
|
assert quota.actions_today(db, user.id, ACTION) == 1
|
|
refused = client.get("/api/search/youtube?q=cats")
|
|
assert refused.status_code == 429
|
|
assert refused.json()["detail"]["reason"] == "search_daily_limit"
|
|
assert _FakeYT.api_calls == 1
|
|
|
|
|
|
def test_an_api_search_the_budget_refused_at_the_call_gives_its_slot_back(
|
|
db, make_user, as_user, pages, monkeypatch
|
|
):
|
|
user = make_user()
|
|
_api(db)
|
|
answers = iter([True, False]) # the pre-check passes; the budget is gone by the call
|
|
monkeypatch.setattr(search_routes.quota, "can_spend", lambda _db, _u: next(answers))
|
|
_FakeYT.api_pages = [_page(1)]
|
|
resp = as_user(user).get("/api/search/youtube?q=cats")
|
|
assert resp.status_code == 429
|
|
assert resp.json()["detail"]["reason"] == "quota_spent"
|
|
assert _FakeYT.api_calls == 0
|
|
assert quota.actions_today(db, user.id, ACTION) == 0
|
|
|
|
|
|
def test_the_budget_check_and_the_call_it_pays_for_hold_one_lock(
|
|
db, make_user, as_user, pages, monkeypatch
|
|
):
|
|
# Checked apart, searches sent together at the last 100 units all passed, and all spent: the
|
|
# check and the call sit inside ONE hold of the lock, not two.
|
|
user = make_user()
|
|
_api(db)
|
|
events: list[str] = []
|
|
real_check = quota.can_spend
|
|
real_call = _FakeYT.search_videos
|
|
|
|
class _Recorded:
|
|
def __init__(self):
|
|
self._lock = threading.Lock()
|
|
|
|
def acquire(self, timeout=-1):
|
|
events.append("acquire")
|
|
return self._lock.acquire(timeout=timeout)
|
|
|
|
def release(self):
|
|
events.append("release")
|
|
self._lock.release()
|
|
|
|
def can_spend(db_, units):
|
|
events.append("check")
|
|
return real_check(db_, units)
|
|
|
|
def search_videos(self, q, page_token=None, order="relevance"):
|
|
events.append("call")
|
|
return real_call(self, q, page_token=page_token, order=order)
|
|
|
|
monkeypatch.setattr(search_routes, "_api_search", _Recorded())
|
|
monkeypatch.setattr(search_routes.quota, "can_spend", can_spend)
|
|
monkeypatch.setattr(_FakeYT, "search_videos", search_videos)
|
|
_FakeYT.api_pages = [_page(1)]
|
|
assert as_user(user).get("/api/search/youtube?q=cats").status_code == 200
|
|
assert events == ["check", "acquire", "check", "call", "release"]
|
|
|
|
|
|
def test_a_search_kept_waiting_for_the_lock_is_refused_as_busy_and_keeps_nothing(
|
|
db, make_user, as_user, pages, monkeypatch
|
|
):
|
|
# Each waiter holds a request worker: a slow YouTube must not park them all behind one another.
|
|
user = make_user()
|
|
_api(db)
|
|
monkeypatch.setattr(search_routes, "API_SEARCH_WAIT_SECONDS", 0.05)
|
|
_FakeYT.api_pages = [_page(1)]
|
|
# Held by another search for a while: a wait without its bound would get it, and be answered.
|
|
assert search_routes._api_search.acquire(timeout=1)
|
|
later = threading.Timer(2.0, search_routes._api_search.release)
|
|
later.start()
|
|
try:
|
|
resp = as_user(user).get("/api/search/youtube?q=cats")
|
|
finally:
|
|
later.join()
|
|
assert resp.status_code == 429
|
|
assert resp.json()["detail"]["reason"] == "search_busy"
|
|
assert _FakeYT.api_calls == 0
|
|
assert quota.actions_today(db, user.id, ACTION) == 0
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"failure",
|
|
[
|
|
YouTubeError("YouTube API unreachable (ConnectError)", reason="transport"),
|
|
YouTubeError("User has no stored OAuth token"),
|
|
],
|
|
ids=["no-connection", "no-token"],
|
|
)
|
|
def test_an_api_search_that_never_reached_youtube_gives_its_slot_back(
|
|
db, make_user, as_user, pages, failure
|
|
):
|
|
user = make_user()
|
|
_api(db)
|
|
sysconfig.set_value(db, "search_daily_limit_per_user", 1)
|
|
_FakeYT.api_pages = [failure, _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
|
|
|
|
|
|
def test_an_api_search_whose_answer_never_came_keeps_its_slot(db, make_user, as_user, pages):
|
|
# The request went out (a read timeout): YouTube may have counted it.
|
|
user = make_user()
|
|
_api(db)
|
|
sysconfig.set_value(db, "search_daily_limit_per_user", 1)
|
|
_FakeYT.api_pages = [
|
|
YouTubeError("YouTube API unreachable (ReadTimeout)", reason="transport", sent=True),
|
|
_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) == 1
|
|
assert client.get("/api/search/youtube?q=cats").status_code == 429
|
|
|
|
|
|
def test_failures_past_the_hours_refunds_keep_their_slot(db, make_user, as_user, pages):
|
|
# A cursor is a free string: a request built to fail would otherwise read YouTube without end.
|
|
user = make_user()
|
|
sysconfig.set_value(db, "search_daily_limit_per_user", 50)
|
|
n = search_routes.REFUNDS_PER_HOUR
|
|
pages["pages"] = [ScrapeError("bad cursor") for _ in range(n + 1)]
|
|
client = as_user(user)
|
|
for _ in range(n + 1):
|
|
assert client.get("/api/search/youtube?q=cats&cursor=bogus").status_code == 422
|
|
assert quota.actions_today(db, user.id, ACTION) == 1
|