tests/e2e/test_privacy_uninstall_e2e.py
Rama dev · réplica de lectura del repositorio original.
"""E2E: row privacy (each user's row hidden from everyone else) and the uninstall that undoes everything.
These are the two promises Shortlist makes that a user cannot verify for themselves: "your rows
are private" and "I can put your server back". Both are checked here against a real server
(the fake PMS/plex.tv), not against mocks that agree with us by construction.
"""
from __future__ import annotations
import re
import time
import pytest
from playwright.sync_api import Page, expect
from tests.e2e.conftest import ShortlistApp, build_real_rows
pytestmark = pytest.mark.e2e
LOAD = 20_000
SLOW = 60_000
class TestRowsStayPrivateAcrossLibraries:
"""The promise of the product, end to end: after a real run, no user can see another
user's row — in ANY library.
This is the shape of the only leak that ever reached a live server (a large production server, 2026-07-12).
Every user's picks were delivered into the movie library regardless of type, so the TV
watchers' rows sat in a movie library holding shows. Plex applies `label!=` share filters
per library, and a collection whose contents don't match its library is matched by neither
filterMovies nor filterTelevision — so those rows were unhidable, and every user saw them.
The excludes really were on the filters the whole time; only looking at a real user's own
Home hubs caught it.
"""
def test_no_user_sees_another_users_row_in_any_library(self, app: ShortlistApp, reset_fake_plex):
state = reset_fake_plex
build_real_rows(app)
owned = {} # slug -> collection ids, from the labels the PMS actually stored
for collection in state.collections.values():
for label in collection.labels:
if label.lower().startswith("shortlist_"):
owned.setdefault(label.lower().removeprefix("shortlist_"), []).append(collection.rating_key)
assert set(owned) == {"sarah", "mike", "jess"}
# sarah watches movies AND TV, so she has a row in each library — the case that leaked.
assert len(owned["sarah"]) == 2, "a both-types watcher must get one row per library"
libraries = {state.collections[key].section_id for key in owned["sarah"]}
assert libraries == {state.section_id, state.show_section_id}
for slug, ids in owned.items():
for key in ids:
collection = state.collections[key]
assert state.filterable(collection), (
f"{slug}'s row in library {collection.section_id} holds items of the wrong type: "
"no share filter can hide it, so every user can see it"
)
# Now look through each user's OWN eyes: their row, and nobody else's.
for account_id, slug in ((201, "sarah"), (202, "mike"), (203, "jess")):
hubs = app.plex_hubs_as(account_id)
visible = {
int(match.group(1))
for hub in hubs
if (match := re.search(r"/library/collections/(\d+)", str(hub.get("key") or "")))
}
assert set(owned[slug]) <= visible, f"{slug} cannot see their own row"
foreign = {key: other for other, ids in owned.items() if other != slug for key in ids}
leaked = {key: foreign[key] for key in visible & set(foreign)}
assert not leaked, f"{slug} can see {sorted(set(leaked.values()))}'s row ({leaked})"
class TestUninstall:
def test_uninstall_puts_the_server_back_as_shortlist_found_it(self, page: Page, app: ShortlistApp, reset_fake_plex):
"""The typed-confirmation path, all the way through: rows deleted, filters restored."""
state = reset_fake_plex
build_real_rows(app)
# 5 rows for 3 users: sarah and the cold-start jess each get one per library; mike watches only TV.
assert len(state.collections) == 5
# People run concurrently, so excludes land in the order their first rows are written; the order
# within one `label!=` clause means nothing to Plex.
clause, _, labels = state.users[201].filters["filterMovies"].partition("!=")
assert (clause, sorted(labels.split(","))) == ("label", ["Shortlist_jess", "Shortlist_mike"])
# Uninstall is its own page now (with a live per-step log), reached from the Danger Zone link.
page.goto("/settings#danger")
page.get_by_role("link", name="Uninstall Shortlist…").click()
expect(page.get_by_role("heading", name="Uninstall Shortlist")).to_be_visible(timeout=LOAD)
commit = page.get_by_role("button", name="Uninstall and restore server")
expect(commit).to_be_disabled()
page.get_by_role("textbox").fill("uninstall shortlist")
expect(commit).to_be_enabled()
commit.click()
expect(page.get_by_text("Uninstall complete")).to_be_visible(timeout=SLOW)
assert state.collections == {}, "a Shortlist collection survived the uninstall"
for user in state.users.values():
assert user.filters["filterMovies"] == "", f"{user.username}'s share filter was not restored"
assert user.filters["filterTelevision"] == ""
def test_uninstall_finishes_when_an_account_has_left_plex(self, page: Page, app: ShortlistApp, reset_fake_plex):
"""Issue #96, reproduced end to end: a user with a snapshot who is gone from plex.tv.
This is the shape a real server reaches by simply un-sharing someone. The restore used to ask
plex.tv for each account as it went, so mike's absence raised `LookupError` out of a loop with
no per-user guard — a 500 that also skipped the collection deletion and row disabling that come
AFTER it, leaving the rows to rebuild on the next scheduled run.
"""
state = reset_fake_plex
build_real_rows(app)
assert len(state.collections) == 5
# mike keeps his snapshot in Shortlist's database, but plex.tv no longer lists his account —
# exactly what removing someone's share does. The user sync is what turns that into a
# recorded departure (`departed_at`), and running the real one here rather than stamping the
# column keeps the whole chain — un-share, sweep, uninstall — under test.
departed = state.users.pop(202)
assert app.api("POST", "/api/users/sync").status_code == 200
# A completed run can still hold the writer lock while its cleanup finishes. The sync
# endpoint then returns queued; wait for the real worker to record the departure before
# testing the confirmed-departure uninstall branch. A missing account alone must not be
# treated as confirmed departed by uninstall.
deadline = time.monotonic() + 120
while time.monotonic() < deadline:
mike = next(user for user in app.api("GET", "/api/users").json() if user["username"] == "mike")
if mike["departed"]:
break
time.sleep(0.25)
assert mike["departed"], "the real sync worker did not record the account's departure"
assert mike["enabled"] is False
page.goto("/settings#danger")
page.get_by_role("link", name="Uninstall Shortlist…").click()
expect(page.get_by_role("heading", name="Uninstall Shortlist")).to_be_visible(timeout=LOAD)
page.get_by_role("textbox").fill("uninstall shortlist")
page.get_by_role("button", name="Uninstall and restore server").click()
# It finishes. Everything downstream of the restore loop is the proof it did not abort.
expect(page.get_by_text("Uninstall complete")).to_be_visible(timeout=SLOW)
assert state.collections == {}, "a Shortlist collection survived an uninstall with a departed user"
# The departed account is NAMED rather than quietly counted as restored, and the page stops
# short of claiming the server is untouched — mike's filters keep our labels for ever.
expect(page.get_by_text("No longer on this server")).to_be_visible()
expect(page.locator("body")).to_contain_text(departed.username)
for user in state.users.values():
assert user.filters["filterMovies"] == "", f"{user.username}'s share filter was not restored"
assert user.filters["filterTelevision"] == ""
def test_uninstall_leaves_collections_shortlist_does_not_own_alone(
self, page: Page, app: ShortlistApp, reset_fake_plex
):
"""Kometa coexistence (plex-safety rule 4): only shortlist_* labelled collections may go."""
from tests.fakes.fake_plex import FakeCollection
state = reset_fake_plex
build_real_rows(app)
foreign = FakeCollection(
rating_key=9999,
title="Kometa: Best of the 90s",
section_id=state.section_id,
labels=["Kometa"],
item_keys=[101, 102],
)
state.collections[9999] = foreign
page.goto("/settings#danger")
page.get_by_role("link", name="Uninstall Shortlist…").click()
expect(page.get_by_role("heading", name="Uninstall Shortlist")).to_be_visible(timeout=LOAD)
# The preview loads on arrival. Its summary counts the 5 Shortlist rows that would go — and
# never the Kometa one.
expect(page.locator("body")).to_contain_text("5 collections", timeout=SLOW)
expect(page.locator("body")).not_to_contain_text("Kometa")
page.get_by_role("textbox").fill("uninstall shortlist")
page.get_by_role("button", name="Uninstall and restore server").click()
expect(page.get_by_text("Uninstall complete")).to_be_visible(timeout=SLOW)
assert list(state.collections) == [9999], "uninstall deleted a collection Shortlist did not create"
assert state.collections[9999].item_keys == [101, 102]
class TestEveryAccountOnTheServerIsCovered:
"""Shortlist's promise is that a user's row is private — from EVERYONE, not from the handful of
accounts Shortlist happens to manage.
On a live server this was not true: 45 of its 48 accounts had completely empty share filters
and could see all three managed users' private rows, because Shortlist only ever wrote filters
for the users it built rows for (a large production server, 2026-07-12).
"""
def test_an_account_shortlist_has_never_seen_still_gets_the_excludes(self, app: ShortlistApp, reset_fake_plex):
"""The owner invites someone to Plex and never opens the Users page. The nightly run must
still stop them seeing everyone else's rows — and must record what it changed on their
share (plex-safety rule 10)."""
from tests.fakes.fake_plex import FakeUser
state = reset_fake_plex
# A stranger: on the Plex server, absent from Shortlist's database entirely.
state.users[299] = FakeUser(id=299, username="stranger")
build_real_rows(app)
# Their share filter now excludes every row that isn't theirs...
stranger = state.users[299]
assert "Shortlist_sarah" in stranger.filters["filterMovies"]
assert "Shortlist_sarah" in stranger.filters["filterTelevision"]
assert "Shortlist_mike" in stranger.filters["filterMovies"]
# ...they see none of those rows on their own Home...
hubs = app.plex_hubs_as(299)
visible = {
int(match.group(1))
for hub in hubs
if (match := re.search(r"/library/collections/(\d+)", str(hub.get("key") or "")))
}
assert not visible, f"a stranger can see {len(visible)} of other people's rows"
# ...and the change to their share is on the record, with the before/after.
# (/api/events is the live SSE stream; /api/events/log is the audit table.)
events = app.api("GET", "/api/events/log?scope=run.privacy_sync").json()
writes = [e for e in events if e["message"]["username"] == "stranger"]
assert writes, "editing someone's Plex share permissions must never go unaudited"
fields = writes[0]["message"]["fields"]
assert fields["filterMovies"]["before"] == ""
assert "Shortlist_sarah" in fields["filterMovies"]["after"]
def test_a_completed_run_is_visible_on_the_run_page(self, page: Page, app: ShortlistApp, reset_fake_plex):
""" "What changed on whose share at 03:31" must be answerable from the UI, not just the
database — so every real run has its own page."""
run = build_real_rows(app)
page.goto(f"/runs/{run['id']}")
expect(page.get_by_role("heading", name=f"Run #{run['id']}")).to_be_visible(timeout=LOAD)