test(plex): pin the late-checkpoint race, its window, and the bulk mark's push supersede

backend/tests/test_plex_finish.py

@@ -13,7 +13,7 @@ from datetime import datetime, timedelta, timezone
 from sqlalchemy import select
 
 from app.db import SessionLocal
-from app.plex.models import PlexItem, PlexLibrary, PlexState
+from app.plex.models import PlexItem, PlexLibrary, PlexSeason, PlexShow, PlexState
 from app.plex import service as plex_service
 from app.routes.plex import playback
 
@@ -138,3 +138,93 @@ def test_marking_watched_supersedes_a_resume_push_still_in_flight(
     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