"""S2 watch-sync semantics: Plex is the source of truth. Two independent pieces: - `_pull_apply` (DB lane): on a genuine divergence for an item Plex knows about, Plex now WINS — the old "keep Siftlode when it's newer within a tolerance" tug-of-war is gone. The anti-bounce short-circuit (`_same_state` + `synced_to_plex`) must still keep a freshly-pushed row from bouncing. - `_should_push_resume` (pure): the in-memory write-back throttle — a `final` checkpoint always pushes, a periodic one pushes at most once per _PUSH_THROTTLE_S seconds per (user, item). """ from datetime import datetime, timedelta, timezone from sqlalchemy import select from app.models import PlexItem, PlexLibrary, PlexState, User from app.plex.watch_sync import _pull_apply from app.routes.plex.playback import ( _PUSH_THROTTLE_S, _resume_push_superseded, _should_push_resume, ) def _seed_item_and_user(db): 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) user = User(email="watch@example.com") db.add(user) db.flush() return item.id, user.id def test_pull_plex_wins_over_diverging_synced_local_resume(db): """The core S2 change: a SYNCED local row (its state already reached Plex) that later diverges from Plex = a real external Plex-side change, so the incremental pull resolves to PLEX's position — Plex is authoritative, no LWW. (A DIRTY local row is protected instead — see the dirty-guard test.)""" item_id, user_id = _seed_item_and_user(db) now = datetime.now(timezone.utc) db.add( PlexState( user_id=user_id, item_id=item_id, status="new", position_seconds=300, progress_updated_at=now, # local touched just now — "newer" than Plex synced_to_plex=True, # but already pushed → a later Plex divergence is an external change ) ) db.flush() # Plex reports a DIFFERENT (older) resume position for the same item. changed = _pull_apply(db, user_id, item_id, "new", 120, plex_ts=now - timedelta(seconds=60)) db.flush() assert changed == "in_progress" st = ( db.execute( select(PlexState).where(PlexState.user_id == user_id, PlexState.item_id == item_id) ) .scalars() .one() ) assert st.position_seconds == 120, "Plex's position must win, even though local was newer" assert st.synced_to_plex is True def test_pull_plex_watched_wins_over_synced_local_in_progress(db): """Plex marking the item watched propagates over a SYNCED local in-progress row (external change).""" item_id, user_id = _seed_item_and_user(db) now = datetime.now(timezone.utc) db.add( PlexState( user_id=user_id, item_id=item_id, status="new", position_seconds=300, progress_updated_at=now, synced_to_plex=True, ) ) db.flush() changed = _pull_apply(db, user_id, item_id, "watched", 0, plex_ts=now - timedelta(seconds=60)) db.flush() assert changed == "watched" st = ( db.execute( select(PlexState).where(PlexState.user_id == user_id, PlexState.item_id == item_id) ) .scalars() .one() ) assert st.status == "watched" assert st.position_seconds == 0 def test_pull_no_bounce_on_matching_row(db): """Anti-bounce short-circuit: when the row already matches what Plex reports (a just-pushed resume), the pull makes NO change and only settles the synced flag — no churn, no re-pull.""" item_id, user_id = _seed_item_and_user(db) now = datetime.now(timezone.utc) db.add( PlexState( user_id=user_id, item_id=item_id, status="new", position_seconds=100, progress_updated_at=now, synced_to_plex=True, # freshly pushed ) ) db.flush() # Plex echoes the same position a beat later (well within the _PROGRESS_MIN_S fuzz). changed = _pull_apply(db, user_id, item_id, "new", 102, plex_ts=now + timedelta(seconds=2)) db.flush() assert changed is None, "a matching row must not bounce" st = ( db.execute( select(PlexState).where(PlexState.user_id == user_id, PlexState.item_id == item_id) ) .scalars() .one() ) assert st.position_seconds == 100, "no churn to the stored position" def test_pull_keeps_dirty_local_on_divergence(db): """Durability guard: a DIRTY row (a local change whose immediate push failed → synced_to_plex=False) must NOT be overwritten by a diverging Plex signal, and must stay dirty so `_repush_dirty` can push the local value up. Overwriting-and-clearing here would strand the local change forever.""" item_id, user_id = _seed_item_and_user(db) now = datetime.now(timezone.utc) db.add( PlexState( user_id=user_id, item_id=item_id, status="watched", # user just marked watched locally; the scrobble to Plex failed position_seconds=0, watched_at=now, progress_updated_at=now, synced_to_plex=False, # dirty — never reached Plex ) ) db.flush() # Plex still lists it in on_deck at an old in-progress offset (it never got the watched scrobble). changed = _pull_apply(db, user_id, item_id, "new", 240, plex_ts=now - timedelta(seconds=60)) db.flush() assert changed is None, "a dirty local row must not be clobbered by the pull" st = ( db.execute( select(PlexState).where(PlexState.user_id == user_id, PlexState.item_id == item_id) ) .scalars() .one() ) assert st.status == "watched", "local watched preserved" assert st.position_seconds == 0 assert st.synced_to_plex is False, "stays dirty so _repush_dirty retries the push" def test_pull_settles_flag_on_matching_dirty_row(db): """Anti-bounce, dirty branch: a row that already MATCHES Plex but is still dirty gets its synced_to_plex flag settled to True (no churn), so `_repush_dirty` stops re-pushing it forever.""" item_id, user_id = _seed_item_and_user(db) now = datetime.now(timezone.utc) db.add( PlexState( user_id=user_id, item_id=item_id, status="new", position_seconds=100, progress_updated_at=now, synced_to_plex=False, # dirty, but already agrees with Plex ) ) db.flush() changed = _pull_apply(db, user_id, item_id, "new", 102, plex_ts=now + timedelta(seconds=2)) db.flush() assert changed is None, "a matching row makes no change" st = ( db.execute( select(PlexState).where(PlexState.user_id == user_id, PlexState.item_id == item_id) ) .scalars() .one() ) assert st.position_seconds == 100, "no churn to the stored position" assert st.synced_to_plex is True, "the dirty flag is settled so it isn't re-pushed endlessly" def test_should_push_resume_scopes_by_user_and_item(): """The throttle window is keyed by the FULL (user, item) tuple: pushing for one key must not suppress a distinct user or a distinct item within the same window.""" now = datetime.now(timezone.utc) assert _should_push_resume(9101, 9201, now, final=False) is True # (u1, i1) first tick # Same window, different item for the same user → independent, still pushes. assert _should_push_resume(9101, 9202, now, final=False) is True # Same window, different user, same item → independent, still pushes. assert _should_push_resume(9102, 9201, now, final=False) is True # And the original key is still throttled within its window (proves the above weren't just "always"). assert _should_push_resume(9101, 9201, now + timedelta(seconds=1), final=False) is False def test_resume_push_superseded_drops_older_pending_push(): """The write-ordering guard: an older still-pending resume push is dropped once a newer push for the same (user, item) has been scheduled, so a reordered stale position can't land at Plex last.""" key = (9301, 9301) t0 = datetime.now(timezone.utc) # Schedule an initial push at t0 (records t0 as the last-push time for this key). assert _should_push_resume(*key, t0, final=False) is True # Nothing newer yet → the t0 push is current, not superseded. assert _resume_push_superseded(*key, t0) is False # A newer push (e.g. a backward seek) is scheduled at t1 > t0. t1 = t0 + timedelta(seconds=1) assert _should_push_resume(*key, t1, final=True) is True # Now the still-pending t0 push is superseded and must drop; the t1 push is current. assert _resume_push_superseded(*key, t0) is True assert _resume_push_superseded(*key, t1) is False def test_should_push_resume_final_always_pushes(): """A final checkpoint (pause / seek / stop) always pushes, bypassing the throttle.""" now = datetime.now(timezone.utc) key = (9001, 9001) assert _should_push_resume(*key, now, final=True) is True # A second final a fraction of a second later still pushes. assert _should_push_resume(*key, now + timedelta(seconds=1), final=True) is True def test_should_push_resume_periodic_is_throttled(): """Periodic (non-final) checkpoints push at most once per _PUSH_THROTTLE_S per (user, item).""" now = datetime.now(timezone.utc) key = (9002, 9002) assert _should_push_resume(*key, now, final=False) is True # first tick pushes # A tick within the window is suppressed. assert ( _should_push_resume(*key, now + timedelta(seconds=_PUSH_THROTTLE_S - 1), final=False) is False ) # Once the window elapses, the next tick pushes again. assert _should_push_resume(*key, now + timedelta(seconds=_PUSH_THROTTLE_S), final=False) is True