231 lines
9.1 KiB
Python
231 lines
9.1 KiB
Python
"""The near-end checkpoint that finishes a Plex item.
|
|
|
|
- Exactly one scrobble per finish, even when two near-end checkpoints race (the 10 s tick and the
|
|
debounced `seeked` flush arrive together): the transition is decided by a conditional write, not a
|
|
read taken before it.
|
|
- The finish supersedes every resume push still in flight, so none can land at Plex after the
|
|
scrobble and revive a resume point on a watched item.
|
|
"""
|
|
|
|
import threading
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import select
|
|
|
|
from app.db import SessionLocal
|
|
from app.plex.models import PlexItem, PlexLibrary, PlexSeason, PlexShow, PlexState
|
|
from app.plex import service as plex_service
|
|
from app.routes.plex import playback
|
|
|
|
|
|
def _item(db) -> PlexItem:
|
|
lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True)
|
|
db.add(lib)
|
|
db.flush()
|
|
item = PlexItem(rating_key="m1", library_id=lib.id, kind="movie", title="A Movie")
|
|
db.add(item)
|
|
db.commit()
|
|
return item
|
|
|
|
|
|
def test_two_racing_finishes_flip_the_row_once(db, make_user):
|
|
user = make_user()
|
|
item = _item(db)
|
|
db.add(PlexState(user_id=user.id, item_id=item.id, status="new", position_seconds=5000))
|
|
db.commit()
|
|
now = datetime.now(timezone.utc)
|
|
|
|
first, second = SessionLocal(), SessionLocal()
|
|
try:
|
|
assert (
|
|
plex_service.mark_finished(first, user.id, item.id, now) is True
|
|
) # holds the row lock
|
|
out: list[bool] = []
|
|
racer = threading.Thread(
|
|
target=lambda: out.append(plex_service.mark_finished(second, user.id, item.id, now))
|
|
)
|
|
racer.start()
|
|
racer.join(0.5)
|
|
assert racer.is_alive(), "the second writer must wait on the first one's row lock"
|
|
first.commit()
|
|
racer.join(5)
|
|
second.commit()
|
|
finally:
|
|
first.close()
|
|
second.close()
|
|
assert out == [False], "the second checkpoint saw the committed finish: no second scrobble"
|
|
|
|
|
|
def test_a_first_ever_finish_inserts_and_counts_once(db, make_user):
|
|
user = make_user()
|
|
item = _item(db)
|
|
now = datetime.now(timezone.utc)
|
|
assert plex_service.mark_finished(db, user.id, item.id, now) is True
|
|
db.commit()
|
|
assert plex_service.mark_finished(db, user.id, item.id, now + timedelta(seconds=10)) is False
|
|
db.commit()
|
|
st = db.execute(select(PlexState).where(PlexState.item_id == item.id)).scalars().one()
|
|
assert (st.status, st.position_seconds) == ("watched", 0)
|
|
|
|
|
|
def test_the_finish_supersedes_a_resume_push_still_in_flight(db, as_user, make_user, monkeypatch):
|
|
user = make_user()
|
|
item = _item(db)
|
|
monkeypatch.setattr(playback.plex_stream, "touch_session", lambda uid, key: None)
|
|
earlier = datetime.now(timezone.utc) - timedelta(seconds=5)
|
|
# A resume push was scheduled a moment ago and has not reached Plex yet.
|
|
assert playback._should_push_resume(user.id, item.id, earlier, final=True)
|
|
|
|
r = as_user(user).post(
|
|
"/api/plex/item/m1/progress", json={"position_seconds": 2995, "duration_seconds": 3000}
|
|
)
|
|
assert r.json() == {"ok": True, "status": "watched", "position_seconds": 0}
|
|
assert playback._resume_push_superseded(user.id, item.id, earlier) is True
|
|
|
|
|
|
def test_a_checkpoint_sent_before_the_finish_does_not_land_on_the_watched_row(
|
|
db, as_user, make_user, monkeypatch
|
|
):
|
|
"""The 10 s tick posted at 2699 of 3000 s (not near the end) races the credits auto-skip's
|
|
"watched". When the mark commits first, the late checkpoint used to write its position onto the
|
|
watched row and schedule a resume push after the scrobble: Plex showed the item in progress, and
|
|
a Plex-authoritative pull imported it back."""
|
|
user = make_user()
|
|
item = _item(db)
|
|
monkeypatch.setattr(playback.plex_stream, "touch_session", lambda uid, key: None)
|
|
client = as_user(user)
|
|
assert client.post("/api/plex/item/m1/state", json={"status": "watched"}).status_code == 200
|
|
|
|
r = client.post(
|
|
"/api/plex/item/m1/progress", json={"position_seconds": 2699, "duration_seconds": 3000}
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
assert r.json() == {"ok": True, "status": "watched", "position_seconds": 0}
|
|
db.expire_all()
|
|
st = db.execute(select(PlexState).where(PlexState.item_id == item.id)).scalars().one()
|
|
assert (st.status, st.position_seconds) == ("watched", 0)
|
|
|
|
|
|
def test_a_rewatch_records_its_position_once_the_finish_is_behind_it(
|
|
db, as_user, make_user, monkeypatch
|
|
):
|
|
"""The other half: a watched item played again is still checkpointed — only a position arriving
|
|
right on the heels of the mark is read as that mark's late tick."""
|
|
user = make_user()
|
|
item = _item(db)
|
|
long_ago = datetime.now(timezone.utc) - timedelta(days=3)
|
|
db.add(PlexState(user_id=user.id, item_id=item.id, status="watched", watched_at=long_ago))
|
|
db.commit()
|
|
monkeypatch.setattr(playback.plex_stream, "touch_session", lambda uid, key: None)
|
|
r = as_user(user).post(
|
|
"/api/plex/item/m1/progress", json={"position_seconds": 600, "duration_seconds": 3000}
|
|
)
|
|
assert r.json() == {"ok": True, "position_seconds": 600}
|
|
db.expire_all()
|
|
st = db.execute(select(PlexState).where(PlexState.item_id == item.id)).scalars().one()
|
|
assert (st.status, st.position_seconds) == ("watched", 600)
|
|
|
|
|
|
def test_marking_watched_supersedes_a_resume_push_still_in_flight(
|
|
db, as_user, make_user, monkeypatch
|
|
):
|
|
"""The same guard the near-end finish has: a resume push scheduled before the mark must not reach
|
|
Plex after its scrobble."""
|
|
user = make_user()
|
|
item = _item(db)
|
|
earlier = datetime.now(timezone.utc) - timedelta(seconds=5)
|
|
assert playback._should_push_resume(user.id, item.id, earlier, final=True)
|
|
r = as_user(user).post("/api/plex/item/m1/state", json={"status": "watched"})
|
|
assert r.status_code == 200, r.text
|
|
assert playback._resume_push_superseded(user.id, item.id, earlier) is True
|
|
|
|
|
|
def _state(db, user, item):
|
|
db.expire_all()
|
|
st = (
|
|
db.execute(
|
|
select(PlexState).where(PlexState.user_id == user.id, PlexState.item_id == item.id)
|
|
)
|
|
.scalars()
|
|
.one()
|
|
)
|
|
return st.status, st.position_seconds
|
|
|
|
|
|
def test_a_mark_committed_after_the_checkpoints_read_still_refuses_it(db, make_user):
|
|
"""The race itself, not its sequential shadow: the route read the row while it was still in
|
|
progress, and the finish committed from another session before the checkpoint wrote. Only a write
|
|
conditional on the row as it is NOW refuses that position."""
|
|
user = make_user()
|
|
item = _item(db)
|
|
db.add(PlexState(user_id=user.id, item_id=item.id, status="new", position_seconds=2690))
|
|
db.commit()
|
|
now = datetime.now(timezone.utc)
|
|
read = plex_service.state_row(db, user, item.id) # what the checkpoint's route read: "new"
|
|
assert read.status == "new"
|
|
|
|
other = SessionLocal()
|
|
try:
|
|
assert plex_service.mark_finished(other, user.id, item.id, now) is True
|
|
other.commit()
|
|
finally:
|
|
other.close()
|
|
|
|
assert plex_service.checkpoint(db, user, item.id, read, 2699, now, dirty=True) is False
|
|
db.commit()
|
|
assert _state(db, user, item) == ("watched", 0)
|
|
|
|
|
|
def test_a_tick_seconds_behind_the_mark_is_refused_a_rewatch_a_minute_later_is_not(db, make_user):
|
|
"""A position right behind the mark is that mark's late tick; one a minute later is a rewatch.
|
|
Literal times on purpose: a window widened to an hour must fail here, not move with it."""
|
|
user = make_user()
|
|
item = _item(db)
|
|
now = datetime.now(timezone.utc)
|
|
st = PlexState(user_id=user.id, item_id=item.id, status="watched", watched_at=now)
|
|
db.add(st)
|
|
db.commit()
|
|
|
|
st.watched_at = now - timedelta(seconds=20)
|
|
db.commit()
|
|
assert plex_service.checkpoint(db, user, item.id, st, 600, now, dirty=True) is False
|
|
db.commit()
|
|
st.watched_at = now - timedelta(seconds=60)
|
|
db.commit()
|
|
assert plex_service.checkpoint(db, user, item.id, st, 600, now, dirty=True) is True
|
|
db.commit()
|
|
assert _state(db, user, item) == ("watched", 600)
|
|
|
|
|
|
def test_marking_a_season_watched_supersedes_its_episodes_resume_pushes(db, as_user, make_user):
|
|
"""The bulk mark has the same guard as one item's: an episode's resume push scheduled before
|
|
the season was marked must not reach Plex after the scrobble."""
|
|
user = make_user()
|
|
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()
|
|
ep = PlexItem(
|
|
rating_key="e1",
|
|
library_id=lib.id,
|
|
kind="episode",
|
|
title="E1",
|
|
show_id=sh.id,
|
|
season_id=se.id,
|
|
season_number=1,
|
|
episode_number=1,
|
|
)
|
|
db.add(ep)
|
|
db.commit()
|
|
earlier = datetime.now(timezone.utc) - timedelta(seconds=5)
|
|
assert playback._should_push_resume(user.id, ep.id, earlier, final=True)
|
|
|
|
r = as_user(user).post("/api/plex/season/se1/state", json={"watched": True})
|
|
assert r.json() == {"ok": True, "changed": 1, "watched": True}
|
|
assert playback._resume_push_superseded(user.id, ep.id, earlier) is True
|