peter
686c673fb2
The player shows what YouTube lists beside the video, without what the user has watched or hidden. Wide, a Related tab beside the description; stacked, a button beside the info opens them in a drawer. A row plays in the same player (Back returns) and the list stays the opened video's. A switch picks on-topic or Hungarian-trending lists, kept on the account. Each related row says whether the user follows its channel; the row is the video card's narrow form with its whole menu. VideoMenu moves to features/video, where the player can reach it.
318 lines
12 KiB
Python
318 lines
12 KiB
Python
"""`GET /api/search/related/{video_id}` — a video's related videos, without what the user has seen.
|
|
|
|
The scrape is replaced by pages the test hands out, and YouTube's videos.list by an offline enrich
|
|
(30-minute videos, so no Shorts probe goes out). What is held: the list leaves out what this user
|
|
watched or hid — here or in their Takeout history, unless taken back to "new" since — blocked
|
|
channels and the video itself; it pages on until it has enough; it lands in the catalog the way
|
|
search finds do; and a failure, or the day's cap, says so instead of answering an empty list."""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from app import quota, sysconfig
|
|
from app.catalog.models import BlockedChannel, Channel, Subscription, Video
|
|
from app.models import QuotaEvent
|
|
from app.reader.models import TakeoutItem, TakeoutKind, VideoState, VideoUnmark
|
|
from app.routes import search as search_routes
|
|
from app.youtube.innertube import ScrapeError
|
|
|
|
T0 = datetime(2026, 10, 1, tzinfo=timezone.utc)
|
|
SOURCE = "srcvideo001"
|
|
|
|
|
|
def _vid(n: int) -> str:
|
|
return f"rel{n:08d}"
|
|
|
|
|
|
def _item(n: int, channel: str = "UCrelated01") -> dict:
|
|
return {
|
|
"id": _vid(n),
|
|
"channel_id": channel,
|
|
"channel_title": f"Chan {channel}",
|
|
"title": f"Related {n}",
|
|
"published_at": None,
|
|
"thumbnail_url": None,
|
|
"live_broadcast": "none",
|
|
}
|
|
|
|
|
|
class _FakeYT:
|
|
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):
|
|
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, recording each call."""
|
|
sysconfig.set_value(db, "shorts_probe_max_seconds", 60)
|
|
state: dict = {"pages": [], "calls": []}
|
|
|
|
def related(video_id, continuation=None, *, hl="en", gl="US"):
|
|
state["calls"].append((video_id, continuation, hl, gl))
|
|
page = state["pages"].pop(0)
|
|
if isinstance(page, Exception):
|
|
raise page
|
|
return page
|
|
|
|
monkeypatch.setattr(search_routes.related_scrape, "related", related)
|
|
monkeypatch.setattr(search_routes, "YouTubeClient", _FakeYT)
|
|
return state
|
|
|
|
|
|
def _page(items: list[dict], token: str | None = "NEXT") -> dict:
|
|
return {"items": items, "next_page_token": token}
|
|
|
|
|
|
def _ids(resp) -> list[str]:
|
|
assert resp.status_code == 200, resp.text
|
|
return [it["id"] for it in resp.json()["items"]]
|
|
|
|
|
|
def _known(db, *ns: int, channel: str = "UCrelated01") -> None:
|
|
"""These related videos already in the catalog (a state needs the row)."""
|
|
if db.get(Channel, channel) is None:
|
|
db.add(Channel(id=channel, title="Chan"))
|
|
for n in ns:
|
|
db.add(Video(id=_vid(n), channel_id=channel, title=f"Related {n}", published_at=T0))
|
|
db.commit()
|
|
|
|
|
|
def test_a_related_list_comes_into_the_catalog_as_search_finds_do(db, make_user, as_user, pages):
|
|
user = make_user()
|
|
pages["pages"] = [_page([_item(n) for n in range(1, 21)])]
|
|
resp = as_user(user).get(f"/api/search/related/{SOURCE}")
|
|
assert _ids(resp) == [_vid(n) for n in range(1, 21)]
|
|
assert resp.json()["next_cursor"] == "NEXT" and resp.json()["has_more"] is True
|
|
video = db.get(Video, _vid(1))
|
|
# Flagged like a search find: hidden from the shared Library, reclaimed when nobody keeps it.
|
|
assert video.via_search is True and video.enriched_at is not None
|
|
assert pages["calls"] == [(SOURCE, None, "en", "US")]
|
|
|
|
|
|
def test_what_the_user_watched_or_hid_here_is_left_out(db, make_user, as_user, pages):
|
|
user = make_user()
|
|
other = make_user()
|
|
_known(db, 1, 2, 3, 4)
|
|
db.add_all(
|
|
[
|
|
VideoState(user_id=user.id, video_id=_vid(1), status="watched", watched_at=T0),
|
|
VideoState(user_id=user.id, video_id=_vid(2), status="hidden"),
|
|
# In progress is not "seen": it stays, to be finished.
|
|
VideoState(user_id=user.id, video_id=_vid(3), status="new", position_seconds=120),
|
|
# Someone else's watch is theirs.
|
|
VideoState(user_id=other.id, video_id=_vid(4), status="watched", watched_at=T0),
|
|
]
|
|
)
|
|
db.commit()
|
|
pages["pages"] = [_page([_item(n) for n in range(1, 6)], token=None)]
|
|
assert _ids(as_user(user).get(f"/api/search/related/{SOURCE}")) == [_vid(3), _vid(4), _vid(5)]
|
|
|
|
|
|
def test_what_the_takeout_history_shows_watched_is_left_out_unless_taken_back(
|
|
db, make_user, as_user, pages
|
|
):
|
|
user = make_user()
|
|
other = make_user()
|
|
_known(db, 2)
|
|
db.add_all(
|
|
[
|
|
# Not in the catalog at all: the history alone names it.
|
|
TakeoutItem(user_id=user.id, kind=TakeoutKind.WATCHED, video_id=_vid(1), at=T0),
|
|
# Watched on YouTube, then taken back to "new" here: it may come again.
|
|
TakeoutItem(user_id=user.id, kind=TakeoutKind.WATCHED, video_id=_vid(2), at=T0),
|
|
VideoUnmark(user_id=user.id, video_id=_vid(2), at=T0 + timedelta(days=1)),
|
|
# Only in Watch later: not watched.
|
|
TakeoutItem(user_id=user.id, kind=TakeoutKind.WATCH_LATER, video_id=_vid(3), at=T0),
|
|
# Another user's history.
|
|
TakeoutItem(user_id=other.id, kind=TakeoutKind.WATCHED, video_id=_vid(4), at=T0),
|
|
]
|
|
)
|
|
db.commit()
|
|
pages["pages"] = [_page([_item(n) for n in range(1, 5)], token=None)]
|
|
assert _ids(as_user(user).get(f"/api/search/related/{SOURCE}")) == [_vid(2), _vid(3), _vid(4)]
|
|
|
|
|
|
def test_an_unmark_older_than_the_history_row_does_not_bring_it_back(db, make_user, as_user, pages):
|
|
user = make_user()
|
|
_known(db, 1)
|
|
db.add_all(
|
|
[
|
|
VideoUnmark(user_id=user.id, video_id=_vid(1), at=T0 - timedelta(days=1)),
|
|
TakeoutItem(user_id=user.id, kind=TakeoutKind.WATCHED, video_id=_vid(1), at=T0),
|
|
]
|
|
)
|
|
db.commit()
|
|
pages["pages"] = [_page([_item(1), _item(2)], token=None)]
|
|
assert _ids(as_user(user).get(f"/api/search/related/{SOURCE}")) == [_vid(2)]
|
|
|
|
|
|
def test_another_users_unmark_does_not_bring_it_back(db, make_user, as_user, pages):
|
|
user = make_user()
|
|
other = make_user()
|
|
_known(db, 1)
|
|
db.add_all(
|
|
[
|
|
TakeoutItem(user_id=user.id, kind=TakeoutKind.WATCHED, video_id=_vid(1), at=T0),
|
|
VideoUnmark(user_id=other.id, video_id=_vid(1), at=T0 + timedelta(days=1)),
|
|
]
|
|
)
|
|
db.commit()
|
|
pages["pages"] = [_page([_item(1), _item(2)], token=None)]
|
|
assert _ids(as_user(user).get(f"/api/search/related/{SOURCE}")) == [_vid(2)]
|
|
|
|
|
|
def test_blocked_channels_and_the_video_itself_are_left_out_and_never_stored(
|
|
db, make_user, as_user, pages
|
|
):
|
|
user = make_user()
|
|
db.add(Channel(id="UCblocked01", title="Blocked"))
|
|
db.commit()
|
|
db.add(BlockedChannel(user_id=user.id, channel_id="UCblocked01"))
|
|
db.commit()
|
|
own = {**_item(9), "id": SOURCE}
|
|
pages["pages"] = [_page([_item(1), _item(2, "UCblocked01"), own], token=None)]
|
|
assert _ids(as_user(user).get(f"/api/search/related/{SOURCE}")) == [_vid(1)]
|
|
assert db.get(Video, _vid(2)) is None
|
|
|
|
|
|
def test_it_pages_on_until_enough_unseen_videos_are_gathered(db, make_user, as_user, pages):
|
|
user = make_user()
|
|
_known(db, *range(1, 16))
|
|
for n in range(1, 16):
|
|
db.add(VideoState(user_id=user.id, video_id=_vid(n), status="watched", watched_at=T0))
|
|
db.commit()
|
|
pages["pages"] = [
|
|
_page([_item(n) for n in range(1, 21)], token="page2"), # 15 seen, 5 new
|
|
_page([_item(n) for n in range(21, 41)], token="page3"), # 20 new: 25 now
|
|
_page([_item(n) for n in range(41, 61)], token="page4"), # never asked
|
|
]
|
|
ids = _ids(resp := as_user(user).get(f"/api/search/related/{SOURCE}"))
|
|
assert ids == [_vid(n) for n in range(16, 41)]
|
|
assert [c[1] for c in pages["calls"]] == [None, "page2"]
|
|
assert resp.json()["next_cursor"] == "page3"
|
|
|
|
|
|
def test_a_repeat_on_a_later_page_counts_once(db, make_user, as_user, pages):
|
|
user = make_user()
|
|
pages["pages"] = [
|
|
_page([_item(n) for n in range(1, 11)], token="page2"),
|
|
_page([_item(n) for n in range(5, 25)], token=None),
|
|
]
|
|
assert _ids(as_user(user).get(f"/api/search/related/{SOURCE}")) == [
|
|
_vid(n) for n in range(1, 25)
|
|
]
|
|
|
|
|
|
def test_it_reads_at_most_four_pages(db, make_user, as_user, pages):
|
|
user = make_user()
|
|
pages["pages"] = [_page([_item(n)], token=f"page{n + 1}") for n in range(1, 7)]
|
|
ids = _ids(resp := as_user(user).get(f"/api/search/related/{SOURCE}"))
|
|
assert ids == [_vid(n) for n in range(1, 5)]
|
|
assert len(pages["calls"]) == 4 and resp.json()["next_cursor"] == "page5"
|
|
|
|
|
|
def test_the_cursor_and_the_hungarian_locale_reach_the_scrape(db, make_user, as_user, pages):
|
|
user = make_user()
|
|
pages["pages"] = [_page([_item(1)], token=None)]
|
|
resp = as_user(user).get(f"/api/search/related/{SOURCE}?cursor=C1®ion=hu")
|
|
assert _ids(resp) == [_vid(1)]
|
|
assert pages["calls"] == [(SOURCE, "C1", "hu", "HU")]
|
|
assert resp.json()["has_more"] is False
|
|
|
|
|
|
def test_an_unknown_region_is_refused(db, make_user, as_user, pages):
|
|
user = make_user()
|
|
assert as_user(user).get(f"/api/search/related/{SOURCE}?region=xx").status_code == 422
|
|
assert pages["calls"] == []
|
|
|
|
|
|
def test_a_failed_scrape_says_youtube_is_unavailable(db, make_user, as_user, pages):
|
|
user = make_user()
|
|
pages["pages"] = [ScrapeError("blocked")]
|
|
resp = as_user(user).get(f"/api/search/related/{SOURCE}")
|
|
assert resp.status_code == 422
|
|
assert resp.json()["detail"] == {"code": "youtube", "reason": "unavailable"}
|
|
|
|
|
|
def test_a_failure_on_a_later_page_keeps_what_came_before(db, make_user, as_user, pages):
|
|
user = make_user()
|
|
pages["pages"] = [_page([_item(1), _item(2)], token="page2"), ScrapeError("blocked")]
|
|
assert _ids(as_user(user).get(f"/api/search/related/{SOURCE}")) == [_vid(1), _vid(2)]
|
|
|
|
|
|
def test_one_list_is_one_event_against_the_days_cap(db, make_user, as_user, pages):
|
|
user = make_user()
|
|
sysconfig.set_value(db, "related_daily_limit_per_user", 1)
|
|
pages["pages"] = [
|
|
_page([_item(1)], token="page2"),
|
|
_page([_item(2)], token=None),
|
|
]
|
|
client = as_user(user)
|
|
assert client.get(f"/api/search/related/{SOURCE}").status_code == 200
|
|
events = db.scalars(
|
|
select(QuotaEvent).where(
|
|
QuotaEvent.user_id == user.id, QuotaEvent.action == quota.QuotaAction.VIDEOS_RELATED
|
|
)
|
|
).all()
|
|
assert len(events) == 1
|
|
refused = client.get(f"/api/search/related/{SOURCE}")
|
|
assert refused.status_code == 429
|
|
assert refused.json()["detail"] == {
|
|
"code": "refused",
|
|
"reason": "related_daily_limit",
|
|
"limit": 1,
|
|
}
|
|
assert len(pages["calls"]) == 2 # the refused one asked YouTube nothing
|
|
|
|
|
|
def test_reading_further_is_the_same_list_not_another_one(db, make_user, as_user, pages):
|
|
user = make_user()
|
|
sysconfig.set_value(db, "related_daily_limit_per_user", 2)
|
|
pages["pages"] = [_page([_item(1)], token=None), _page([_item(2)], token=None)]
|
|
client = as_user(user)
|
|
assert client.get(f"/api/search/related/{SOURCE}").status_code == 200
|
|
assert client.get(f"/api/search/related/{SOURCE}?cursor=more").status_code == 200
|
|
assert quota.actions_today(db, user.id, quota.QuotaAction.VIDEOS_RELATED) == 1
|
|
|
|
|
|
def test_each_row_says_whether_the_user_follows_its_channel(db, make_user, as_user, pages):
|
|
user = make_user()
|
|
stranger = make_user()
|
|
_known(db, 1, channel="UCfollowed1")
|
|
_known(db, 2, channel="UCstranger1")
|
|
db.add(Subscription(user_id=user.id, channel_id="UCfollowed1"))
|
|
db.add(Subscription(user_id=stranger.id, channel_id="UCstranger1"))
|
|
db.commit()
|
|
pages["pages"] = [
|
|
_page([_item(1, "UCfollowed1"), _item(2, "UCstranger1"), _item(3, "UCnobody01")], None)
|
|
]
|
|
rows = as_user(user).get(f"/api/search/related/{SOURCE}").json()["items"]
|
|
assert {r["id"]: r["followed"] for r in rows} == {
|
|
_vid(1): True,
|
|
_vid(2): False, # another user's follow is not this one's
|
|
_vid(3): False,
|
|
}
|