peter
533571f572
PlexLibrary, PlexShow, PlexSeason, PlexItem, PlexPerson, PlexCollection, PlexState, PlexLink, PlexPlaylist and PlexPlaylistItem leave app/models.py for app/plex/models.py; app.tables registers them. Every import is re-pointed; no schema change.
279 lines
11 KiB
Python
279 lines
11 KiB
Python
"""The daily full watch reconcile (`run_plex_watch_reconcile`) — the one pass that DELETES watch state:
|
|
a mirrored row Plex no longer has a record for is read as "Plex un-watched it" and cleared. A review
|
|
found it trusting a scan a Plex hiccup had cut short; these hold what it may and may not clear.
|
|
"""
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from app import sysconfig
|
|
from app.models import User
|
|
from app.plex.models import PlexItem, PlexLibrary, PlexLink, PlexState
|
|
from app.plex import watch_sync
|
|
|
|
|
|
class _FakePlex:
|
|
"""A PlexClient stand-in: one movie section whose listing reports `total` (more than it hands
|
|
over = a walk cut short), plus any `more` sections as `(type, leaves, total)` — listed AFTER
|
|
the movies — and a record of every write the reconcile sends back to Plex."""
|
|
|
|
def __init__(self, leaves: list[dict], total: int | None = None, more=()):
|
|
self.listings = {"1": ("movie", leaves, len(leaves) if total is None else total)}
|
|
for i, (stype, extra, extra_total) in enumerate(more, start=2):
|
|
self.listings[str(i)] = (stype, extra, extra_total)
|
|
self.calls: list[tuple] = []
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
return False
|
|
|
|
def sections(self):
|
|
return [{"type": stype, "key": key} for key, (stype, _l, _t) in self.listings.items()]
|
|
|
|
def section_items(self, key, item_type, start, size):
|
|
_stype, leaves, total = self.listings[key]
|
|
return leaves[start : start + size], total
|
|
|
|
def scrobble(self, rk):
|
|
self.calls.append(("scrobble", rk))
|
|
|
|
def unscrobble(self, rk):
|
|
self.calls.append(("unscrobble", rk))
|
|
|
|
def set_timeline(self, rk, time_ms, duration_ms):
|
|
self.calls.append(("timeline", rk, time_ms))
|
|
|
|
|
|
@pytest.fixture
|
|
def world(db):
|
|
"""Five mirrored movies and the owner's link; each test sets the local states it needs."""
|
|
sysconfig.set_value(db, "plex_enabled", True)
|
|
lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True)
|
|
user = User(email="reconcile@example.com")
|
|
db.add_all([lib, user])
|
|
db.flush()
|
|
items = {}
|
|
for rk in ("m1", "m2", "m3", "m4", "m5"):
|
|
items[rk] = PlexItem(
|
|
rating_key=rk, library_id=lib.id, kind="movie", title=rk, duration_s=6000
|
|
)
|
|
db.add(items[rk])
|
|
db.add(
|
|
PlexLink(
|
|
user_id=user.id,
|
|
uses_admin=True,
|
|
sync_enabled=True,
|
|
initial_import_done=True,
|
|
plex_account_id=1,
|
|
)
|
|
)
|
|
db.flush()
|
|
|
|
def state(rk, **kw):
|
|
db.add(PlexState(user_id=user.id, item_id=items[rk].id, **kw))
|
|
|
|
def rows():
|
|
db.expire_all()
|
|
by_id = {i.id: rk for rk, i in items.items()}
|
|
return {
|
|
by_id[s.item_id]: (s.status, s.position_seconds, s.synced_to_plex)
|
|
for s in db.execute(select(PlexState).where(PlexState.user_id == user.id)).scalars()
|
|
if s.item_id in by_id # the five movies; a test's own extra items it reads itself
|
|
}
|
|
|
|
return state, rows
|
|
|
|
|
|
def _run(db, monkeypatch, plex):
|
|
monkeypatch.setattr(watch_sync, "PlexClient", lambda _db: plex)
|
|
return watch_sync.run_plex_watch_reconcile(db)
|
|
|
|
|
|
def _local_states(state):
|
|
state("m1", status="watched", synced_to_plex=True) # mirrored; Plex has no record now
|
|
state("m2", status="watched", synced_to_plex=False) # local change that never reached Plex
|
|
state("m3", status="hidden", synced_to_plex=True) # Siftlode-only, never cleared
|
|
state("m5", status="new", position_seconds=300, synced_to_plex=True) # Plex says otherwise
|
|
|
|
|
|
def test_a_complete_scan_clears_what_plex_unwatched_and_keeps_the_rest(db, monkeypatch, world):
|
|
state, rows = world
|
|
_local_states(state)
|
|
db.commit()
|
|
plex = _FakePlex(
|
|
[
|
|
{"ratingKey": "m3"}, # listed, no watch signal
|
|
{"ratingKey": "m4", "viewCount": 1, "lastViewedAt": 1_700_000_000},
|
|
{"ratingKey": "m5", "viewOffset": 120_000, "lastViewedAt": 1_700_000_000},
|
|
]
|
|
)
|
|
out = _run(db, monkeypatch, plex)
|
|
assert rows() == {
|
|
"m2": ("watched", 0, True), # re-pushed, then settled
|
|
"m3": ("hidden", 0, True),
|
|
"m4": ("watched", 0, True), # pulled in
|
|
"m5": ("new", 120, True), # Plex's position wins
|
|
}
|
|
assert ("scrobble", "m2") in plex.calls
|
|
assert out["cleared"] == 1 and out["repushed"] == 1 and "incomplete" not in out
|
|
|
|
|
|
def test_a_scan_cut_short_clears_nothing(db, monkeypatch, world):
|
|
state, rows = world
|
|
_local_states(state)
|
|
db.commit()
|
|
# The server said 5, the walk got 3: m1 may simply not have been reached.
|
|
plex = _FakePlex(
|
|
[
|
|
{"ratingKey": "m3"},
|
|
{"ratingKey": "m4", "viewCount": 1, "lastViewedAt": 1_700_000_000},
|
|
{"ratingKey": "m5", "viewOffset": 120_000, "lastViewedAt": 1_700_000_000},
|
|
],
|
|
total=5,
|
|
)
|
|
out = _run(db, monkeypatch, plex)
|
|
after = rows()
|
|
assert after["m1"] == ("watched", 0, True) # NOT un-watched on a partial picture
|
|
assert after["m4"] == ("watched", 0, True) and after["m5"] == ("new", 120, True) # pulls stand
|
|
assert after["m2"] == ("watched", 0, True) # the dirty row is still pushed up
|
|
assert out["cleared"] == 0 and out["incomplete"] == 1
|
|
|
|
|
|
_LISTED = [
|
|
{"ratingKey": "m3"},
|
|
{"ratingKey": "m4", "viewCount": 1, "lastViewedAt": 1_700_000_000},
|
|
{"ratingKey": "m5", "viewOffset": 120_000, "lastViewedAt": 1_700_000_000},
|
|
]
|
|
|
|
|
|
def test_a_genuinely_empty_second_section_still_lets_the_reconcile_clear(db, monkeypatch, world):
|
|
# The server SAYS the shows section holds 0: that is a complete listing, not a hiccup — before
|
|
# a review it read as "incomplete" forever and un-watches never synced again.
|
|
state, rows = world
|
|
_local_states(state)
|
|
db.commit()
|
|
out = _run(db, monkeypatch, _FakePlex(_LISTED, more=[("show", [], 0)]))
|
|
assert "m1" not in rows()
|
|
assert out["cleared"] == 1 and "incomplete" not in out
|
|
|
|
|
|
def _watched_episode_in_a_shows_section(db) -> int:
|
|
"""A second mirrored section (key "2") holding one episode the owner has watched, synced."""
|
|
shows = PlexLibrary(plex_key="2", title="Shows", kind="show", enabled=True)
|
|
db.add(shows)
|
|
db.flush()
|
|
ep = PlexItem(rating_key="e1", library_id=shows.id, kind="episode", title="e1")
|
|
db.add(ep)
|
|
db.flush()
|
|
user_id = db.execute(select(PlexLink.user_id)).scalar_one()
|
|
db.add(PlexState(user_id=user_id, item_id=ep.id, status="watched", synced_to_plex=True))
|
|
return ep.id
|
|
|
|
|
|
def _state_of(db, item_id: int) -> str | None:
|
|
db.expire_all()
|
|
st = db.execute(select(PlexState).where(PlexState.item_id == item_id)).scalar_one_or_none()
|
|
return st.status if st else None
|
|
|
|
|
|
def test_a_section_we_mirror_that_says_0_keeps_its_rows_and_does_not_hold_up_the_rest(
|
|
db, monkeypatch, world
|
|
):
|
|
# A total of 0 from a section our items come from is a Plex that lost the library for a moment,
|
|
# not one that un-watched all of it: its rows stay. And it is decided per
|
|
# section: at first one such section (a library emptied on Plex, its mirrored items kept)
|
|
# made EVERY run incomplete, and un-watches in every other section never synced again.
|
|
state, rows = world
|
|
_local_states(state)
|
|
ep_id = _watched_episode_in_a_shows_section(db)
|
|
db.commit()
|
|
out = _run(db, monkeypatch, _FakePlex(_LISTED, more=[("show", [], 0)]))
|
|
assert _state_of(db, ep_id) == "watched"
|
|
assert "m1" not in rows() # the movies section was walked whole: Plex un-watched m1
|
|
assert out["cleared"] == 1 and out["incomplete"] == 1
|
|
|
|
|
|
def test_a_library_the_allowlist_leaves_out_keeps_its_watch_states(db, monkeypatch, world):
|
|
# Its items are absent from every scan because the scan never walks it — not because Plex
|
|
# un-watched them. It cleared them on every run.
|
|
state, _rows = world
|
|
other = PlexLibrary(plex_key="2", title="Shows", kind="show", enabled=True)
|
|
db.add(other)
|
|
db.flush()
|
|
ep = PlexItem(rating_key="e1", library_id=other.id, kind="episode", title="e1")
|
|
db.add(ep)
|
|
db.flush()
|
|
user_id = db.execute(select(PlexLink.user_id)).scalar_one()
|
|
db.add(PlexState(user_id=user_id, item_id=ep.id, status="watched", synced_to_plex=True))
|
|
_local_states(state)
|
|
sysconfig.set_value(db, "plex_libraries", "1")
|
|
db.commit()
|
|
out = _run(db, monkeypatch, _FakePlex(_LISTED, more=[("show", [], 0)]))
|
|
db.expire_all()
|
|
left = {
|
|
rk
|
|
for rk in db.execute(
|
|
select(PlexItem.rating_key).join(PlexState, PlexState.item_id == PlexItem.id)
|
|
).scalars()
|
|
}
|
|
assert "m1" not in left # the walked section still clears
|
|
kept = db.execute(select(PlexState).where(PlexState.item_id == ep.id)).scalar_one_or_none()
|
|
assert kept is not None and kept.status == "watched"
|
|
assert out["cleared"] == 1
|
|
|
|
|
|
def test_a_cut_short_section_is_not_rescued_by_a_complete_one_after_it(db, monkeypatch, world):
|
|
# Each section settles on its own walk: movies cut short (5 said, 3 handed), then a complete
|
|
# shows listing — m1 may simply not have been reached.
|
|
state, rows = world
|
|
_local_states(state)
|
|
db.commit()
|
|
shows = [{"ratingKey": "s1"}]
|
|
out = _run(db, monkeypatch, _FakePlex(_LISTED, total=5, more=[("show", shows, 1)]))
|
|
assert rows()["m1"] == ("watched", 0, True)
|
|
assert out["cleared"] == 0 and out["incomplete"] == 1
|
|
|
|
|
|
def test_a_listing_without_a_total_is_never_complete(db, monkeypatch, world):
|
|
state, rows = world
|
|
_local_states(state)
|
|
ep_id = _watched_episode_in_a_shows_section(db)
|
|
db.commit()
|
|
out = _run(db, monkeypatch, _FakePlex(_LISTED, more=[("show", [], None)]))
|
|
assert _state_of(db, ep_id) == "watched" # its section never said how much it holds
|
|
assert "m1" not in rows()
|
|
assert out["cleared"] == 1 and out["incomplete"] == 1
|
|
|
|
|
|
def test_an_empty_listing_is_not_a_library_of_unwatched_titles(db, monkeypatch, world):
|
|
state, rows = world
|
|
state("m1", status="watched", synced_to_plex=True)
|
|
db.commit()
|
|
out = _run(db, monkeypatch, _FakePlex([]))
|
|
assert rows() == {"m1": ("watched", 0, True)}
|
|
assert out["cleared"] == 0
|
|
|
|
|
|
def test_a_dirty_resume_is_re_pushed_as_a_timeline_and_a_dirty_reset_as_an_unscrobble(
|
|
db, monkeypatch, world
|
|
):
|
|
state, rows = world
|
|
state("m1", status="new", position_seconds=300, synced_to_plex=False)
|
|
state("m2", status="new", position_seconds=2, synced_to_plex=False) # under the 5 s floor
|
|
db.commit()
|
|
plex = _FakePlex([{"ratingKey": rk} for rk in ("m1", "m2", "m3", "m4", "m5")])
|
|
_run(db, monkeypatch, plex)
|
|
assert ("timeline", "m1", 300_000) in plex.calls
|
|
assert ("unscrobble", "m2") in plex.calls
|
|
assert {rk: synced for rk, (_s, _p, synced) in rows().items()} == {"m1": True, "m2": True}
|
|
|
|
|
|
def test_watch_fields_map_to_a_state_only_when_plex_holds_a_signal():
|
|
to = watch_sync._plex_watch_to_state
|
|
assert to(None, None, None) is None
|
|
assert to(0, 4_000, None) is None # a stray few seconds is not "in progress"
|
|
assert to(0, 90_000, 1_700_000_000)[:2] == ("new", 90)
|
|
assert to("2", 90_000, 1_700_000_000)[:2] == ("watched", 0) # watched wins over an offset
|