peter
bc8fe2cc6c
A subscription pulled in from YouTube left an explored channel flagged as an exploration's, so its later videos came in hidden from the Library with their watched marks waiting for good; a subscribe landing while an exploration's videos came in left those flagged too. One step, settle_explored, now holds the rule by who follows the channel now: the subscribe here, the subscription import, the watched-channel videos and the explore page all end with it. Migration 0116 clears what the gap left behind. The list's "may still grow" count no longer counts a channel it already shows from a stub row, and the comments on the cleanup's keep rule and on the in-memory daily shares say what they do.
382 lines
14 KiB
Python
382 lines
14 KiB
Python
"""Catalog's service and queries, one use case each, below the HTTP layer.
|
|
|
|
The routes are thin callers now, and their own tests stay the contract a client sees. These hold
|
|
what each write does and leaves on its own: that it flushes and leaves the commit to its caller,
|
|
that a row of someone else's reads as a missing one, and the rule each write carries (an explored
|
|
channel flagged only while nobody follows it, a subscription that keeps it, a cleared search that
|
|
spares what anyone holds). DB-backed (real Postgres).
|
|
"""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import select
|
|
|
|
from app.catalog import queries, search, service
|
|
from app.db import SessionLocal
|
|
from app.catalog.models import (
|
|
BlockedChannel,
|
|
Channel,
|
|
ExploredChannel,
|
|
SearchFind,
|
|
Subscription,
|
|
Video,
|
|
)
|
|
from app.reader import queries as reader_queries
|
|
from app.reader import service as reader_service
|
|
from app.sync import explore as sync_explore
|
|
from app.sync import subscriptions as sync_subscriptions
|
|
|
|
|
|
def _channel(db, cid="UCc", **kw):
|
|
ch = Channel(id=cid, title=kw.pop("title", cid), **kw)
|
|
db.add(ch)
|
|
db.commit()
|
|
return ch
|
|
|
|
|
|
def _video(db, vid, cid="UCc", **kw):
|
|
v = Video(id=vid, channel_id=cid, published_at=datetime.now(timezone.utc), **kw)
|
|
db.add(v)
|
|
db.commit()
|
|
return v
|
|
|
|
|
|
def _sub(db, user, cid="UCc", **kw):
|
|
s = Subscription(user_id=user.id, channel_id=cid, **kw)
|
|
db.add(s)
|
|
db.commit()
|
|
return s
|
|
|
|
|
|
def _uncommitted(db):
|
|
return bool(db.new or db.dirty or db.deleted) or db.in_transaction()
|
|
|
|
|
|
# --- whose a row is ------------------------------------------------------------------------
|
|
|
|
|
|
def test_a_users_subscriptions_blocks_and_explorations_are_theirs_alone(db, make_user):
|
|
owner, stranger = make_user(), make_user()
|
|
ch = _channel(db)
|
|
_sub(db, owner)
|
|
service.explore(db, owner, ch)
|
|
service.block(db, owner, "UCc")
|
|
db.commit()
|
|
assert queries.subscription(db, owner, "UCc") is not None
|
|
assert queries.subscription(db, stranger, "UCc") is None
|
|
assert queries.exploration(db, stranger, "UCc") is None
|
|
assert queries.is_blocked(db, stranger, "UCc") is False
|
|
assert queries.blocked_ids(db, stranger) == set()
|
|
assert [s.channel_id for s, _ in queries.subscriptions(db, stranger, "en")] == []
|
|
|
|
|
|
# --- subscription settings -------------------------------------------------------------------
|
|
|
|
|
|
def test_full_history_reports_only_the_switch_on(db, make_user):
|
|
user = make_user()
|
|
_channel(db)
|
|
sub = _sub(db, user)
|
|
assert service.update_subscription(db, sub, priority=3) is False
|
|
assert service.update_subscription(db, sub, deep_requested=True) is True
|
|
assert service.update_subscription(db, sub, deep_requested=True) is False # already on
|
|
assert _uncommitted(db)
|
|
db.commit()
|
|
assert (sub.priority, sub.hidden, sub.deep_requested) == (3, False, True)
|
|
|
|
|
|
def test_full_history_for_all_moves_only_the_callers_rows(db, make_user):
|
|
user, other = make_user(), make_user()
|
|
_channel(db)
|
|
_sub(db, user)
|
|
theirs = _sub(db, other)
|
|
assert service.set_deep_all(db, user, True) == 1
|
|
db.commit()
|
|
db.expire_all()
|
|
assert theirs.deep_requested is False
|
|
|
|
|
|
def test_a_reset_asks_every_subscriber_for_full_history(db, make_user):
|
|
a, b = make_user(), make_user()
|
|
ch = _channel(db, backfill_done=True, recent_synced_at=datetime.now(timezone.utc))
|
|
_sub(db, a)
|
|
_sub(db, b)
|
|
service.reset_backfill(db, ch)
|
|
db.commit()
|
|
db.expire_all()
|
|
assert (ch.backfill_done, ch.recent_synced_at) == (False, None)
|
|
assert set(db.scalars(select(Subscription.deep_requested))) == {True}
|
|
|
|
|
|
# --- explore, subscribe, block ---------------------------------------------------------------
|
|
|
|
|
|
def test_an_exploration_is_one_row_and_flags_the_channel_only_while_nobody_follows_it(
|
|
db, make_user
|
|
):
|
|
explorer, follower = make_user(), make_user()
|
|
lone = _channel(db, "UClone")
|
|
kept = _channel(db, "UCkept")
|
|
_sub(db, follower, "UCkept")
|
|
service.explore(db, explorer, lone)
|
|
service.explore(db, explorer, kept)
|
|
db.commit()
|
|
first = queries.exploration(db, explorer, "UClone").created_at
|
|
service.explore(db, explorer, lone) # refreshes the grace clock
|
|
db.commit()
|
|
assert db.scalar(select(ExploredChannel.id).where(ExploredChannel.channel_id == "UClone"))
|
|
assert len(db.scalars(select(ExploredChannel.id)).all()) == 2
|
|
assert queries.exploration(db, explorer, "UClone").created_at >= first
|
|
assert (lone.from_explore, kept.from_explore) == (True, False)
|
|
|
|
|
|
def test_subscribing_keeps_an_explored_channel_and_its_videos(db, make_user):
|
|
user = make_user()
|
|
ch = _channel(db, from_explore=True)
|
|
v = _video(db, "v1", via_explore=True)
|
|
service.record_subscription(db, user, ch, "")
|
|
db.commit()
|
|
db.expire_all()
|
|
sub = queries.subscription(db, user, "UCc")
|
|
assert (sub.yt_subscription_id, ch.from_explore, v.via_explore) == (None, False, False)
|
|
service.drop_subscription(db, sub)
|
|
db.commit()
|
|
assert queries.subscription(db, user, "UCc") is None
|
|
|
|
|
|
def test_exploring_a_channel_someone_follows_brings_its_uploads_in_as_the_catalogs_own(
|
|
db, make_user, monkeypatch
|
|
):
|
|
follower, explorer = make_user(), make_user()
|
|
ch = _channel(db, uploads_playlist_id="UUc")
|
|
db.add(Subscription(user_id=follower.id, channel_id="UCc"))
|
|
db.commit()
|
|
service.explore(db, explorer, ch)
|
|
db.commit()
|
|
|
|
class _Yt:
|
|
def get_playlist_items(self, playlist_id, page_token):
|
|
item = {"contentDetails": {"videoId": "v1", "videoPublishedAt": "2026-05-01T00:00:00Z"}}
|
|
return {"items": [item]}
|
|
|
|
monkeypatch.setattr(sync_explore, "_enrich_new", lambda session, yt, ids: 0)
|
|
sync_explore.explore_ingest_page(db, _Yt(), ch)
|
|
db.commit()
|
|
db.expire_all()
|
|
# The channel is followed, so not the exploration's: a video flagged as one would stay flagged
|
|
# (subscribing clears only an explored channel's) and its watched marks would wait for good.
|
|
assert (ch.from_explore, db.get(Video, "v1").via_explore) == (False, False)
|
|
|
|
|
|
def test_a_follow_pulled_in_from_youtube_makes_an_explored_channel_the_catalogs_own(
|
|
db, make_user, monkeypatch
|
|
):
|
|
explorer, follower = make_user(), make_user()
|
|
ch = _channel(db, details_synced_at=datetime.now(timezone.utc))
|
|
service.explore(db, explorer, ch)
|
|
db.commit()
|
|
_video(db, "v1", via_explore=True)
|
|
|
|
class _Yt:
|
|
def __init__(self, session, user):
|
|
pass
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
return False
|
|
|
|
def iter_subscriptions(self):
|
|
yield {"channel_id": "UCc", "title": "UCc", "yt_subscription_id": "sub-UCc"}
|
|
|
|
def get_channels(self, ids):
|
|
return []
|
|
|
|
monkeypatch.setattr(sync_subscriptions, "YouTubeClient", _Yt)
|
|
sync_subscriptions.import_subscriptions(db, follower)
|
|
db.commit()
|
|
db.expire_all()
|
|
# Followed on youtube.com is followed all the same: left flagged, the channel's later videos
|
|
# came in as an exploration's, hidden from the Library and their watched marks waiting for good.
|
|
assert (db.get(Channel, "UCc").from_explore, db.get(Video, "v1").via_explore) == (False, False)
|
|
|
|
|
|
def test_a_follow_landing_while_an_explored_page_comes_in_leaves_none_of_it_flagged(
|
|
db, make_user, monkeypatch
|
|
):
|
|
explorer, follower = make_user(), make_user()
|
|
ch = _channel(db, uploads_playlist_id="UUc")
|
|
service.explore(db, explorer, ch)
|
|
db.commit()
|
|
|
|
class _Yt:
|
|
def get_playlist_items(self, playlist_id, page_token):
|
|
# Someone subscribes, in a request of their own, while YouTube answers this one.
|
|
with SessionLocal() as other:
|
|
service.record_subscription(other, follower, other.get(Channel, "UCc"), "sub-UCc")
|
|
other.commit()
|
|
item = {"contentDetails": {"videoId": "v1", "videoPublishedAt": "2026-05-01T00:00:00Z"}}
|
|
return {"items": [item]}
|
|
|
|
monkeypatch.setattr(sync_explore, "_enrich_new", lambda session, yt, ids: 0)
|
|
sync_explore.explore_ingest_page(db, _Yt(), ch)
|
|
db.commit()
|
|
db.expire_all()
|
|
# The subscribe cleared what was flagged before it; a page that lands after it must not be.
|
|
assert (db.get(Channel, "UCc").from_explore, db.get(Video, "v1").via_explore) == (False, False)
|
|
|
|
|
|
def test_a_block_is_once_and_ends_the_exploration(db, make_user):
|
|
user = make_user()
|
|
ch = _channel(db)
|
|
service.explore(db, user, ch)
|
|
db.commit()
|
|
service.block(db, user, "UCc")
|
|
service.block(db, user, "UCc")
|
|
db.commit()
|
|
assert len(db.scalars(select(BlockedChannel.id)).all()) == 1
|
|
assert queries.exploration(db, user, "UCc") is None
|
|
assert [c.id for c in queries.blocked_channels(db, user, "en")] == ["UCc"]
|
|
service.unblock(db, user, "UCc")
|
|
db.commit()
|
|
assert queries.is_blocked(db, user, "UCc") is False
|
|
|
|
|
|
# --- what the status cards count -------------------------------------------------------------
|
|
|
|
|
|
def test_my_progress_counts_only_my_channels_and_my_unhidden_videos(db, make_user):
|
|
user, other = make_user(), make_user()
|
|
now = datetime.now(timezone.utc)
|
|
_channel(db, "UCa", details_synced_at=now, recent_synced_at=now, backfill_done=True)
|
|
_channel(db, "UCb")
|
|
_channel(db, "UCtheirs")
|
|
_video(db, "a1", "UCa")
|
|
_video(db, "b1", "UCb")
|
|
_video(db, "t1", "UCtheirs")
|
|
_sub(db, user, "UCa", deep_requested=True)
|
|
_sub(db, user, "UCb", hidden=True)
|
|
_sub(db, other, "UCtheirs")
|
|
assert queries.my_progress(db, user) == (2, 1, 1, 1)
|
|
assert [c.id for c in queries.my_deep_requested(db, user)] == ["UCa"]
|
|
assert queries.my_video_count(db, user) == 1
|
|
assert queries.video_total(db) == 3
|
|
assert queries.catalog_totals(db, user)["subscriptions"] == 2
|
|
|
|
|
|
def test_a_channel_with_no_title_is_left_out_of_the_titles(db):
|
|
_channel(db, "UCt", title="Titled")
|
|
db.add(Channel(id="UCn", title=None))
|
|
db.commit()
|
|
assert queries.channel_titles(db, {"UCt", "UCn", "UCgone"}) == {"UCt": "Titled"}
|
|
assert queries.channel_title(db, "UCt") == "Titled"
|
|
|
|
|
|
def test_the_deep_backlog_is_the_unfinished_channels_someone_asked_for(db, make_user):
|
|
user = make_user()
|
|
_channel(db, "UCwant")
|
|
_channel(db, "UCdone", backfill_done=True)
|
|
_channel(db, "UCnobody")
|
|
_sub(db, user, "UCwant", deep_requested=True)
|
|
_sub(db, user, "UCdone", deep_requested=True)
|
|
_sub(db, user, "UCnobody")
|
|
assert [c.id for c in queries.deep_backlog(db)] == ["UCwant"]
|
|
assert queries.deep_queued(db, ["UCwant", "UCdone", "UCnobody"]) == {"UCwant", "UCdone"}
|
|
|
|
|
|
# --- live search -----------------------------------------------------------------------------
|
|
|
|
|
|
def test_a_cleared_search_spares_what_anyone_holds(db, make_user):
|
|
user, other, follower = make_user(), make_user(), make_user()
|
|
_channel(db, "UCs")
|
|
_channel(db, "UCfollowed")
|
|
_sub(db, follower, "UCfollowed")
|
|
for vid in ("loose", "watched", "found_by_other"):
|
|
_video(db, vid, "UCs", via_search=True)
|
|
_video(db, "followed", "UCfollowed", via_search=True)
|
|
_video(db, "organic", "UCs")
|
|
ids = ["loose", "watched", "found_by_other", "followed", "organic"]
|
|
search.record_finds(db, user, ids, "cat")
|
|
search.record_finds(db, user, ids, "cat") # idempotent
|
|
search.record_finds(db, other, ["found_by_other"], "dog")
|
|
reader_service.mark(db, other, "watched", "watched")
|
|
db.commit()
|
|
assert search.clear(db, user, ids) == 1
|
|
db.commit()
|
|
left = set(db.scalars(select(Video.id)))
|
|
assert left == {"watched", "found_by_other", "followed", "organic"}
|
|
assert db.scalars(select(SearchFind.user_id)).all() == [other.id]
|
|
|
|
|
|
def test_a_search_puts_what_the_user_already_has_last(db, make_user):
|
|
user = make_user()
|
|
_channel(db, "UCs")
|
|
_channel(db, "UCmine")
|
|
_sub(db, user, "UCmine")
|
|
for vid, cid in (("new1", "UCs"), ("mine", "UCmine"), ("seen", "UCs"), ("new2", "UCs")):
|
|
_video(db, vid, cid)
|
|
reader_service.mark(db, user, "seen", "hidden")
|
|
db.commit()
|
|
assert search.new_first(db, user, ["mine", "new1", "seen", "new2"]) == [
|
|
"new1",
|
|
"new2",
|
|
"mine",
|
|
"seen",
|
|
]
|
|
|
|
|
|
# --- what Catalog asks Reader ----------------------------------------------------------------
|
|
|
|
|
|
def test_a_users_tag_links_are_made_once_and_only_theirs_are_read(db, make_user):
|
|
user, other = make_user(), make_user()
|
|
_channel(db)
|
|
mine = reader_service.create_tag(db, user, "Mine", "other", None)
|
|
theirs = reader_service.create_tag(db, other, "Theirs", "other", None)
|
|
reader_service.attach_user_tag(db, user, "UCc", mine.id)
|
|
reader_service.attach_user_tag(db, user, "UCc", mine.id)
|
|
reader_service.attach_user_tag(db, other, "UCc", theirs.id)
|
|
db.commit()
|
|
assert reader_queries.user_tag_links(db, user) == {"UCc": [mine.id]}
|
|
assert reader_queries.user_tag_links(db, user, "UCelse") == {}
|
|
reader_service.detach_user_tag(db, user, "UCc", mine.id)
|
|
db.commit()
|
|
assert reader_queries.user_tag_links(db, user) == {}
|
|
assert reader_queries.user_tag_links(db, other) == {"UCc": [theirs.id]}
|
|
|
|
|
|
def test_discovery_offers_the_playlist_channels_not_followed_and_not_ones_own(db, make_user):
|
|
user = make_user()
|
|
for cid in ("UCnew", "UCfollowed", "UCown"):
|
|
_channel(db, cid)
|
|
for vid, cid in (("n1", "UCnew"), ("n2", "UCnew"), ("f1", "UCfollowed"), ("o1", "UCown")):
|
|
_video(db, vid, cid)
|
|
_sub(db, user, "UCfollowed")
|
|
pl = reader_service.create_playlist(db, user, "L")
|
|
for vid in ("n1", "n2", "f1", "o1"):
|
|
reader_service.add_item(db, pl, vid, mark_dirty=False)
|
|
db.commit()
|
|
rows = reader_queries.playlist_channels(db, user, "UCown", "en")
|
|
assert [(ch.id, n, lists) for ch, n, lists in rows] == [("UCnew", 2, 1)]
|
|
|
|
|
|
def test_an_exploration_older_than_its_grace_is_still_one_row(db, make_user):
|
|
"""The grace clock is refreshed in place: re-exploring after the cutoff does not add a row."""
|
|
user = make_user()
|
|
ch = _channel(db)
|
|
db.add(
|
|
ExploredChannel(
|
|
user_id=user.id,
|
|
channel_id="UCc",
|
|
created_at=datetime.now(timezone.utc) - timedelta(days=30),
|
|
)
|
|
)
|
|
db.commit()
|
|
service.explore(db, user, ch)
|
|
db.commit()
|
|
rows = db.scalars(select(ExploredChannel)).all()
|
|
assert len(rows) == 1
|
|
assert rows[0].created_at > datetime.now(timezone.utc) - timedelta(minutes=1)
|