peter
c18a2063c8
A reorder was invisible to the overlap bookkeeping: a failed removal reset the screen from a cache without it, and the membership guard never corrects an order. It now counts as a write, a rollback leaves the order to the read after the last write, and the accepted order keeps the rows it does not name, as the server does. Undo places come from the order on screen. A stored position landing during a write reaches the screen, and what a read brought is taken once the refresh is in. A write that went through drops its rows from the cache again, past a read from before it. Closing the player after Trash took the row it opened from no longer strands the focus: the row in that place takes it, else the header's box, else the list's menu. The page menu's grouping row closes its menu when it goes with the view. One hidden row gets its own wording. /search/clear refuses a body without its list.
411 lines
15 KiB
Python
411 lines
15 KiB
Python
"""Writes driven through HTTP, each read back: the handler called its unit and committed.
|
|
|
|
A route's helper only flushes; the handler commits. The `db` fixture fails a test whose request left
|
|
a write uncommitted, so every test here drives one successful write through its route and then
|
|
reads the result back. The read-back holds that the handler called its unit; it runs in the
|
|
request's own session, so it sees a write that was never committed — the commit is the fixture's
|
|
to hold (a test that must see it committed reads from a session of its own). The services' own
|
|
tests hold what each write does; these hold that the route finishes it. Nothing here reaches
|
|
YouTube or Plex: the YouTube client is a stub and no user has a Plex link. Needs the DB.
|
|
"""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from app import quota
|
|
from app.auth import require_write_scope
|
|
from app.catalog.models import Channel, ExploredChannel, SearchFind, Subscription, Video
|
|
from app.downloads.models import DownloadJob, DownloadProfile, DownloadShare
|
|
from app.plex.models import PlexItem, PlexLibrary, PlexSeason, PlexShow, PlexState
|
|
from app.db import SessionLocal
|
|
from app.models import Notification
|
|
from app.notifications import READ_SOFT_CAP
|
|
from app.reader.models import SavedView, Tag, VideoState
|
|
from app.routes import channels
|
|
|
|
NOW = datetime.now(timezone.utc)
|
|
|
|
|
|
def _video(db, vid="v1", **kw):
|
|
if db.get(Channel, "UCr") is None:
|
|
db.add(Channel(id="UCr", title="R"))
|
|
db.flush()
|
|
db.add(Video(id=vid, channel_id="UCr", published_at=NOW, duration_seconds=60, **kw))
|
|
db.commit()
|
|
|
|
|
|
# --- tags ------------------------------------------------------------------------------------
|
|
|
|
|
|
def test_a_tag_made_through_the_route_is_listed_back(db, make_user, as_user):
|
|
user = make_user()
|
|
client = as_user(user)
|
|
r = client.post("/api/tags", json={"name": "Science"})
|
|
assert r.status_code == 200, r.text
|
|
assert [t["name"] for t in client.get("/api/tags").json()] == ["Science"]
|
|
|
|
|
|
def test_a_tag_renamed_through_the_route_is_listed_back(db, make_user, as_user):
|
|
user = make_user()
|
|
tag = Tag(user_id=user.id, name="Old")
|
|
db.add(tag)
|
|
db.commit()
|
|
client = as_user(user)
|
|
assert client.patch(f"/api/tags/{tag.id}", json={"name": "New"}).status_code == 200
|
|
assert [t["name"] for t in client.get("/api/tags").json()] == ["New"]
|
|
|
|
|
|
def test_a_tag_deleted_through_the_route_is_gone(db, make_user, as_user):
|
|
user = make_user()
|
|
tag = Tag(user_id=user.id, name="Gone")
|
|
db.add(tag)
|
|
db.commit()
|
|
client = as_user(user)
|
|
assert client.delete(f"/api/tags/{tag.id}").json() == {"deleted": tag.id}
|
|
assert client.get("/api/tags").json() == []
|
|
|
|
|
|
# --- watch state ------------------------------------------------------------------------------
|
|
|
|
|
|
def _state(db, user, vid="v1"):
|
|
db.expire_all()
|
|
return db.execute(
|
|
select(VideoState).where(VideoState.user_id == user.id, VideoState.video_id == vid)
|
|
).scalar_one_or_none()
|
|
|
|
|
|
def test_a_video_marked_watched_through_the_route_stays_watched(db, make_user, as_user):
|
|
user = make_user()
|
|
_video(db)
|
|
r = as_user(user).post("/api/videos/v1/state", json={"status": "watched"})
|
|
assert r.status_code == 200, r.text
|
|
assert _state(db, user).status == "watched"
|
|
|
|
|
|
def test_a_video_unmarked_through_the_route_stays_unmarked(db, make_user, as_user):
|
|
# "new" is its own branch of the handler, with its own commit.
|
|
user = make_user()
|
|
_video(db)
|
|
db.add(VideoState(user_id=user.id, video_id="v1", status="watched"))
|
|
db.commit()
|
|
r = as_user(user).post("/api/videos/v1/state", json={"status": "new"})
|
|
assert r.status_code == 200, r.text
|
|
assert _state(db, user) is None
|
|
|
|
|
|
def test_a_video_state_cleared_through_the_route_is_gone(db, make_user, as_user):
|
|
user = make_user()
|
|
_video(db)
|
|
db.add(VideoState(user_id=user.id, video_id="v1", status="hidden", position_seconds=90))
|
|
db.commit()
|
|
assert as_user(user).delete("/api/videos/v1/progress").status_code == 200
|
|
db.expire_all()
|
|
row = _state(db, user)
|
|
assert (row.status, row.position_seconds) == ("hidden", 0)
|
|
|
|
|
|
# --- saved views ------------------------------------------------------------------------------
|
|
|
|
|
|
def test_views_reordered_through_the_route_list_back_in_that_order(db, make_user, as_user):
|
|
user = make_user()
|
|
a = SavedView(user_id=user.id, source="feed", name="A", filters={}, position=0)
|
|
b = SavedView(user_id=user.id, source="feed", name="B", filters={}, position=1)
|
|
db.add_all([a, b])
|
|
db.commit()
|
|
client = as_user(user)
|
|
assert client.post("/api/saved-views/reorder", json={"ids": [b.id, a.id]}).status_code == 200
|
|
listed = client.get("/api/saved-views").json()
|
|
assert [(v["name"], v["position"]) for v in listed] == [("B", 0), ("A", 1)]
|
|
|
|
|
|
# --- download profiles and shares -------------------------------------------------------------
|
|
|
|
_SPEC = {"format": "best"}
|
|
|
|
|
|
def _profile(db, user, name="Mine"):
|
|
p = DownloadProfile(user_id=user.id, name=name, spec=_SPEC, is_builtin=False, sort_order=101)
|
|
db.add(p)
|
|
db.commit()
|
|
return p
|
|
|
|
|
|
def _profile_names(client):
|
|
return [p["name"] for p in client.get("/api/downloads/profiles").json()]
|
|
|
|
|
|
def test_a_profile_made_through_the_route_is_listed_back(db, make_user, as_user):
|
|
user = make_user()
|
|
client = as_user(user)
|
|
r = client.post("/api/downloads/profiles", json={"name": "Phone", "spec": _SPEC})
|
|
assert r.status_code == 200, r.text
|
|
assert _profile_names(client) == ["Phone"]
|
|
|
|
|
|
def test_a_profile_renamed_through_the_route_is_listed_back(db, make_user, as_user):
|
|
user = make_user()
|
|
p = _profile(db, user)
|
|
client = as_user(user)
|
|
r = client.patch(f"/api/downloads/profiles/{p.id}", json={"name": "Renamed"})
|
|
assert r.status_code == 200, r.text
|
|
assert _profile_names(client) == ["Renamed"]
|
|
|
|
|
|
def test_a_profile_deleted_through_the_route_is_gone(db, make_user, as_user):
|
|
user = make_user()
|
|
p = _profile(db, user)
|
|
client = as_user(user)
|
|
assert client.delete(f"/api/downloads/profiles/{p.id}").json() == {"deleted": p.id}
|
|
assert _profile_names(client) == []
|
|
|
|
|
|
def test_a_share_taken_back_through_the_route_is_gone(db, make_user, as_user):
|
|
owner, friend = make_user(), make_user()
|
|
job = DownloadJob(
|
|
user_id=owner.id, source_kind="youtube", source_ref="dQw4w9WgXcQ", profile_snapshot={}
|
|
)
|
|
db.add(job)
|
|
db.flush()
|
|
db.add(DownloadShare(job_id=job.id, from_user_id=owner.id, to_user_id=friend.id))
|
|
db.commit()
|
|
r = as_user(owner).delete(f"/api/downloads/{job.id}/share/{friend.email}")
|
|
assert r.json() == {"ok": True}
|
|
db.expire_all()
|
|
assert db.execute(select(DownloadShare)).first() is None
|
|
|
|
|
|
# --- channels ---------------------------------------------------------------------------------
|
|
|
|
|
|
class _YouTube:
|
|
"""The YouTube client, answering every write the channel routes make."""
|
|
|
|
def __init__(self, db, user):
|
|
pass
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
return False
|
|
|
|
def insert_subscription(self, channel_id):
|
|
return "yt-sub-1"
|
|
|
|
def delete_subscription(self, sub_id):
|
|
return None
|
|
|
|
def get_channels(self, ids):
|
|
return []
|
|
|
|
|
|
@pytest.fixture
|
|
def youtube(monkeypatch):
|
|
monkeypatch.setattr(channels, "YouTubeClient", _YouTube)
|
|
monkeypatch.setattr(quota, "backfill_reserve_reached", lambda: False)
|
|
|
|
|
|
def _channel(db, cid="UCc", **kw):
|
|
ch = Channel(id=cid, title="C", details_synced_at=NOW, **kw)
|
|
db.add(ch)
|
|
db.commit()
|
|
return ch
|
|
|
|
|
|
def _subscription(db, user, cid="UCc"):
|
|
db.expire_all()
|
|
return db.execute(
|
|
select(Subscription).where(Subscription.user_id == user.id, Subscription.channel_id == cid)
|
|
).scalar_one_or_none()
|
|
|
|
|
|
def test_a_subscribe_through_the_route_is_recorded(db, make_user, as_user, override_dep, youtube):
|
|
user = make_user()
|
|
_channel(db)
|
|
client = as_user(user)
|
|
override_dep(require_write_scope, lambda: user)
|
|
assert client.post("/api/channels/UCc/subscribe").json() == {"ok": True, "subscribed": "UCc"}
|
|
assert _subscription(db, user).yt_subscription_id == "yt-sub-1"
|
|
|
|
|
|
def test_an_explore_through_the_route_is_recorded(db, make_user, as_user, youtube, monkeypatch):
|
|
user = make_user()
|
|
_channel(db)
|
|
# The ingest pages YouTube; what the route itself records is the exploration.
|
|
monkeypatch.setattr(channels, "details_stale", lambda ch, now: False)
|
|
monkeypatch.setattr(
|
|
channels,
|
|
"explore_ingest_page",
|
|
lambda db, yt, ch, token: {"ingested": 0, "next_page_token": None},
|
|
)
|
|
r = as_user(user).post("/api/channels/UCc/explore")
|
|
assert r.status_code == 200, r.text
|
|
db.expire_all()
|
|
assert db.execute(select(ExploredChannel.user_id)).scalars().all() == [user.id]
|
|
assert db.get(Channel, "UCc").from_explore is True
|
|
|
|
|
|
def test_an_unsubscribe_through_the_route_drops_the_subscription(
|
|
db, make_user, as_user, override_dep, youtube
|
|
):
|
|
user = make_user()
|
|
_channel(db)
|
|
db.add(Subscription(user_id=user.id, channel_id="UCc", yt_subscription_id="yt-sub-1"))
|
|
db.commit()
|
|
client = as_user(user)
|
|
override_dep(require_write_scope, lambda: user)
|
|
r = client.delete("/api/channels/UCc/subscription")
|
|
assert r.json() == {"ok": True, "unsubscribed": "UCc"}
|
|
assert _subscription(db, user) is None
|
|
|
|
|
|
def test_a_backfill_reset_through_the_route_is_kept(db, make_user, as_user, monkeypatch):
|
|
admin, reader = make_user(role="admin"), make_user()
|
|
_channel(db, backfill_done=True, backfill_cursor="page-9", recent_synced_at=NOW)
|
|
db.add(Subscription(user_id=reader.id, channel_id="UCc", deep_requested=False))
|
|
db.commit()
|
|
# The recent pull afterwards reaches YouTube; the reset is the route's own write.
|
|
monkeypatch.setattr(channels, "run_recent_backfill", lambda *a, **kw: None)
|
|
r = as_user(admin).post("/api/channels/UCc/reset-backfill")
|
|
assert r.status_code == 200, r.text
|
|
db.expire_all()
|
|
ch = db.get(Channel, "UCc")
|
|
assert (ch.backfill_done, ch.backfill_cursor, ch.recent_synced_at) == (False, None, None)
|
|
assert _subscription(db, reader).deep_requested is True
|
|
|
|
|
|
# --- live search ------------------------------------------------------------------------------
|
|
|
|
|
|
def test_search_results_cleared_through_the_route_are_gone(db, make_user, as_user):
|
|
user = make_user()
|
|
_video(db, "s1", via_search=True)
|
|
db.add(SearchFind(user_id=user.id, video_id="s1"))
|
|
db.commit()
|
|
r = as_user(user).post("/api/search/clear", json={"video_ids": ["s1"]})
|
|
assert r.json() == {"ok": True, "removed": 1}
|
|
db.expire_all()
|
|
assert db.get(Video, "s1") is None
|
|
assert db.execute(select(SearchFind)).first() is None
|
|
|
|
|
|
def test_clearing_leaves_a_video_the_caller_never_found(db, make_user, as_user):
|
|
# A related list ingests its rows with no find: they are in someone's list right now, and an id
|
|
# any user could name must not delete them from under it.
|
|
user = make_user()
|
|
_video(db, "s1", via_search=True)
|
|
_video(db, "s2", via_search=True)
|
|
db.add(SearchFind(user_id=user.id, video_id="s1"))
|
|
db.commit()
|
|
r = as_user(user).post("/api/search/clear", json={"video_ids": ["s1", "s2"]})
|
|
assert r.json() == {"ok": True, "removed": 1}
|
|
db.expire_all()
|
|
assert db.get(Video, "s1") is None and db.get(Video, "s2") is not None
|
|
# Naming only ids it never found removes nothing.
|
|
r = as_user(user).post("/api/search/clear", json={"video_ids": ["s2"]})
|
|
assert r.json() == {"ok": True, "removed": 0}
|
|
|
|
|
|
def test_clearing_refuses_a_body_without_its_list(db, make_user, as_user):
|
|
# A misnamed key read as an empty list: 200, and nothing cleared.
|
|
user = make_user()
|
|
r = as_user(user).post("/api/search/clear", json={"ids": ["s1"]})
|
|
assert r.status_code == 422
|
|
assert as_user(user).post("/api/search/clear", json={"video_ids": []}).status_code == 200
|
|
|
|
|
|
# --- Plex show and season marks ---------------------------------------------------------------
|
|
|
|
|
|
def _show(db):
|
|
lib = PlexLibrary(plex_key="2", title="Shows", kind="show", enabled=True)
|
|
db.add(lib)
|
|
db.flush()
|
|
sh = PlexShow(rating_key="sh1", library_id=lib.id, title="Show")
|
|
db.add(sh)
|
|
db.flush()
|
|
se = PlexSeason(rating_key="se1", show_id=sh.id, season_number=1)
|
|
db.add(se)
|
|
db.flush()
|
|
for n in (1, 2):
|
|
db.add(
|
|
PlexItem(
|
|
rating_key=f"e{n}",
|
|
library_id=lib.id,
|
|
kind="episode",
|
|
title=f"E{n}",
|
|
show_id=sh.id,
|
|
season_id=se.id,
|
|
season_number=1,
|
|
episode_number=n,
|
|
)
|
|
)
|
|
db.commit()
|
|
|
|
|
|
def _plex_marks(db, user):
|
|
db.expire_all()
|
|
return sorted(
|
|
db.execute(
|
|
select(PlexItem.rating_key, PlexState.status)
|
|
.join(PlexState, PlexState.item_id == PlexItem.id)
|
|
.where(PlexState.user_id == user.id)
|
|
).all()
|
|
)
|
|
|
|
|
|
def test_a_show_marked_watched_through_the_route_stays_watched(db, make_user, as_user):
|
|
user = make_user()
|
|
_show(db)
|
|
r = as_user(user).post("/api/plex/show/sh1/state", json={"watched": True})
|
|
assert r.json() == {"ok": True, "changed": 2, "watched": True}
|
|
assert _plex_marks(db, user) == [("e1", "watched"), ("e2", "watched")]
|
|
|
|
|
|
def test_a_season_marked_watched_through_the_route_stays_watched(db, make_user, as_user):
|
|
user = make_user()
|
|
_show(db)
|
|
r = as_user(user).post("/api/plex/season/se1/state", json={"watched": True})
|
|
assert r.json() == {"ok": True, "changed": 2, "watched": True}
|
|
assert _plex_marks(db, user) == [("e1", "watched"), ("e2", "watched")]
|
|
|
|
|
|
def _read_past_the_cap(db, user) -> list[Notification]:
|
|
"""READ_SOFT_CAP read notices, oldest first, and one unread notice after them."""
|
|
base = datetime(2024, 1, 1, tzinfo=timezone.utc)
|
|
rows = [
|
|
Notification(
|
|
user_id=user.id,
|
|
type="t",
|
|
title=f"n{i}",
|
|
read=i < READ_SOFT_CAP,
|
|
created_at=base + timedelta(minutes=i),
|
|
)
|
|
for i in range(READ_SOFT_CAP + 1)
|
|
]
|
|
db.add_all(rows)
|
|
db.commit()
|
|
return rows
|
|
|
|
|
|
def _committed_notices(user) -> tuple[int, bool]:
|
|
"""What another session sees: how many of the user's notices are left, and whether the oldest is."""
|
|
with SessionLocal() as s:
|
|
ids = s.scalars(select(Notification.title).where(Notification.user_id == user.id)).all()
|
|
return len(ids), "n0" in ids
|
|
|
|
|
|
@pytest.mark.parametrize("path", ["{id}/read", "read_all"])
|
|
def test_reading_a_notice_past_the_cap_commits_the_trim(db, make_user, as_user, path):
|
|
"""Marking a notice read makes READ_SOFT_CAP + 1 read ones: the handler commits the read flag and
|
|
the trim of the oldest together, and the trim counts the notice just read."""
|
|
user = make_user()
|
|
rows = _read_past_the_cap(db, user)
|
|
r = as_user(user).post("/api/me/notifications/" + path.format(id=rows[-1].id))
|
|
assert r.json() == {"ok": True}
|
|
assert _committed_notices(user) == (READ_SOFT_CAP, False)
|