tests/e2e/test_users_runs_e2e.py
Rama dev · réplica de lectura del repositorio original.
"""E2E: Users, per-user overrides, triggering runs, and reading the result.
The through-line: a control the owner touches in the browser must change what the ENGINE does
on the next run. A toggle that only repaints the UI is the failure this file exists to catch.
"""
from __future__ import annotations
import re
import time
from collections.abc import Callable
import pytest
from playwright.sync_api import Page, expect
from tests.e2e.conftest import ShortlistApp, build_real_rows
from tests.fakes.fake_plex import DEMO_MOVIES, DEMO_SHOWS
pytestmark = pytest.mark.e2e
def open_rows(page) -> None:
"""Open every collapsed row card on a run page.
A person's results live inside the row that built them, and a row with siblings starts closed.
Re-queries each time rather than iterating a snapshot: clicking one row re-renders the list, which
detaches every handle taken before it.
"""
for _ in range(8):
closed = page.get_by_role("button", expanded=False)
if not closed.count():
return
closed.first.click()
LOAD = 20_000
SLOW = 90_000
def _users_by_name(app: ShortlistApp) -> dict[str, dict]:
return {user["username"]: user for user in app.api("GET", "/api/users").json()}
def _wait_until(condition: Callable[[], bool], message: str, timeout_s: float = 10.0) -> None:
"""Poll the API until a persisted value shows up — robust to the auto-save debounce, which the
'Saved' indicator can race (it flips before the PUT the debounce fired has fully committed)."""
deadline = time.monotonic() + timeout_s
while time.monotonic() < deadline:
if condition():
return
time.sleep(0.25)
raise AssertionError(message)
class TestTheOwnerIsAUserToo:
"""The person running Shortlist gets a row like anyone else (issue #1).
plex.tv's `/api/users` returns everyone the server is shared WITH and never its owner, so
before this the owner had no user row at all — and a one-person server built nothing.
"""
def test_a_server_where_only_the_owner_is_enabled_still_builds_a_row(
self, page: Page, app: ShortlistApp, reset_fake_plex
):
"""The solo-server case end to end: sync finds the owner, and a run gives them real picks
from their own watch history — not the popular-titles fallback, which is what an empty
history would silently produce."""
state = reset_fake_plex
app.api("POST", "/api/users/sync")
owner = _users_by_name(app)[state.owner_username]
assert owner["user_type"] == "owner"
assert owner["enabled"] is False # arrives switched off, like every other new user
for user in _users_by_name(app).values():
app.api("PATCH", f"/api/users/{user['id']}", json={"enabled": user["id"] == owner["id"]})
run = build_real_rows(app)
rows = [c for c in state.collections.values() if any(lbl.lower() == "shortlist_steve" for lbl in c.labels)]
assert rows, "the owner is the only enabled user and got no row at all"
for collection in rows:
assert collection.item_keys
assert collection.promoted_own_home, "the owner's row never reached the owner's own Home"
assert run["stats"]["users_ok"] == 1
# Their picks come from THEIR history — PMS files the owner's plays under a local account
# id, so asking by plex.tv id returns nothing and this row would be the cold-start one.
page.goto(f"/users/{owner['id']}")
expect(page.get_by_role("heading", name=state.owner_username)).to_be_visible(timeout=LOAD)
expect(page.get_by_text("New viewer")).to_have_count(0)
def test_the_owners_row_is_excluded_on_every_share_plex_will_accept_one_for(
self, app: ShortlistApp, reset_fake_plex
):
"""The owner's own filter is never written (Plex cannot restrict them) — that must not be
mistaken for 'this row needs no hiding'. Everyone else still has to exclude it.
Everyone Plex ALLOWS, that is. A managed account with a parental profile is the one exception:
plex.tv answers 422 to any label restriction while the profile is set, so no exclude can exist
on that share no matter what Shortlist does. Asserting it anyway would be asserting a write
that cannot succeed. What must hold instead is that the account is REPORTED rather than
silently passed over — which is the rest of this test (#76)."""
state = reset_fake_plex
app.api("POST", "/api/users/sync")
owner = _users_by_name(app)[state.owner_username]
app.api("PATCH", f"/api/users/{owner['id']}", json={"enabled": True})
build_real_rows(app)
unfilterable = {u.id for u in state.users.values() if u.restriction_profile}
assert unfilterable, "the roster must carry a profiled account, or this proves only the easy half"
for account_id, user in state.users.items():
if account_id in unfilterable:
continue
for field_name in ("filterMovies", "filterTelevision"):
assert "Shortlist_steve" in user.filters[field_name], (
f"account {account_id} can see the owner's row in {field_name}"
)
# The exception is not a blind spot: the run looked through that account's own eyes and said
# what it found. A skip that reported nothing is the bug this whole check exists for.
listed = _users_by_name(app)
for account_id in unfilterable:
username = state.users[account_id].username
assert listed[username]["unhidden_rows"] > 0, (
f"{username} was skipped by the filter write AND reported as seeing nothing — "
"the exact silence that hid this for a year"
)
class TestUsers:
def test_selected_pause_survives_reload_without_changing_access_or_other_people(
self, page: Page, app: ShortlistApp
):
before = _users_by_name(app)
page.goto("/users")
page.get_by_label("Search users").fill("sarah")
# The checkboxes appear once selecting is switched on.
page.get_by_role("button", name="Select people", exact=True).click()
page.get_by_role("checkbox", name="Select sarah", exact=True).check()
page.get_by_role("button", name="Pause rebuilding", exact=True).click()
_wait_until(
lambda: _users_by_name(app)["sarah"]["prefs"].get("paused") is True,
"selected pause must reach the stored preferences",
)
page.reload()
sarah = page.get_by_role("row").filter(has=page.get_by_role("link", name="sarah", exact=True))
expect(sarah.get_by_text("Paused", exact=True)).to_be_visible(timeout=LOAD)
paused = _users_by_name(app)
assert {name: user["enabled"] for name, user in paused.items()} == {
name: user["enabled"] for name, user in before.items()
}
for name in before.keys() - {"sarah"}:
assert paused[name]["prefs"] == before[name]["prefs"]
page.get_by_role("button", name="Select people", exact=True).click() # a reload leaves selecting off
page.get_by_role("checkbox", name="Select sarah", exact=True).check()
page.get_by_role("button", name="Resume rebuilding", exact=True).click()
_wait_until(
lambda: _users_by_name(app)["sarah"]["prefs"].get("paused") is False,
"selected resume must reach the stored preferences",
)
assert _users_by_name(app)["sarah"]["enabled"] == before["sarah"]["enabled"]
def test_disabling_a_user_persists_and_leaves_the_others_alone(self, page: Page, app: ShortlistApp):
page.goto("/users")
jess = page.get_by_role("switch", name="Shortlist row for jess")
expect(jess).to_be_checked(timeout=LOAD)
jess.click()
expect(jess).not_to_be_checked(timeout=LOAD)
page.reload()
expect(page.get_by_role("switch", name="Shortlist row for jess")).not_to_be_checked(timeout=LOAD)
# Exactly one user changed — a broadcast PATCH would pass a "it persisted" test too.
users = _users_by_name(app)
assert users["jess"]["enabled"] is False
assert users["sarah"]["enabled"] is True
assert users["mike"]["enabled"] is True
def test_per_user_overrides_persist(self, page: Page, app: ShortlistApp):
sarah_id = _users_by_name(app)["sarah"]["id"]
page.goto(f"/users/{sarah_id}")
expect(page.get_by_role("heading", name="sarah")).to_be_visible(timeout=LOAD)
# Pause is a user-level control in the header (switch is "on" when active).
paused = page.get_by_role("switch", name="Pause or resume sarah")
expect(paused).to_be_checked(timeout=LOAD)
paused.click()
expect(paused).not_to_be_checked(timeout=LOAD)
# Size is a PER-ROW override now: open a row's drawer, switch on a custom size and type 10.
# The drawer AUTO-SAVES (no Save button — the whole app is one save paradigm).
page.get_by_role("button", name="Customize for this person").first.click()
page.get_by_role("switch", name="Custom row size for this person").click()
titles = page.get_by_label("Titles for this person")
titles.fill("10")
titles.blur() # the free number field commits on blur
expect(page.get_by_text("Saved").first).to_be_visible(timeout=LOAD)
# Pause reaches the user's prefs; the size override lives on the row, not user prefs. Poll the
# API for the persisted value rather than trusting the indicator, which races the debounce.
assert _users_by_name(app)["sarah"]["prefs"]["paused"] is True
expect_row_size = 10
def _size_persisted() -> bool:
rows = app.api("GET", f"/api/users/{sarah_id}/rows").json()
return any(r["override"].get("row_size") == expect_row_size for r in rows)
_wait_until(_size_persisted, "the per-row size override must persist")
# Exactly this user changed — mike's prefs stay empty.
assert _users_by_name(app)["mike"]["prefs"] == {}
def test_a_paused_user_is_skipped_by_the_next_run(self, page: Page, app: ShortlistApp, reset_fake_plex):
"""The override has to reach the ENGINE, not just the users table."""
state = reset_fake_plex
sarah_id = _users_by_name(app)["sarah"]["id"]
page.goto(f"/users/{sarah_id}")
expect(page.get_by_role("heading", name="sarah")).to_be_visible(timeout=LOAD)
# The header pause switch is "on" when active; clicking it pauses this user.
paused = page.get_by_role("switch", name="Pause or resume sarah")
expect(paused).to_be_checked(timeout=LOAD)
paused.click()
expect(paused).not_to_be_checked(timeout=LOAD)
run = build_real_rows(app)
built = {result["slug"] for result in app.api("GET", f"/api/runs/{run['id']}").json()["users"]}
assert built == {"mike", "jess"}, "a paused user must not be rebuilt"
labels = {label.lower() for c in state.collections.values() for label in c.labels}
# sarah's OWNER label is the one that must be absent — that is what "not rebuilt" looks like
# on the server. The constant `shortlist` label is on every row we write and names nobody, so
# it is expected here and says nothing about who was built.
assert labels == {"shortlist", "shortlist_mike", "shortlist_jess"}, (
"sarah's OWNER label must be absent — that is what 'not rebuilt' looks like on the server"
)
assert "shortlist_sarah" not in labels
class TestRuns:
def test_a_run_started_from_the_ui_lands_in_runs_with_per_user_rows(
self, page: Page, app: ShortlistApp, reset_fake_plex
):
state = reset_fake_plex
page.goto("/runs")
expect(page.get_by_text("No run history")).to_be_visible(timeout=LOAD)
page.get_by_role("button", name="Run all rows now").click()
# The row appears the moment the run is queued — the owner is never left guessing.
run_link = page.get_by_role("link", name="#1")
expect(run_link).to_be_visible(timeout=LOAD)
expect(page.get_by_role("cell", name="manual")).to_be_visible()
app.wait_for_run(1)
page.reload()
# The result cell leads with the title-cased status ("OK · 1 warning") — scoped to the table,
# since the page's stats bar also shows "OK" as the last-run status. The users cell keeps "3 ok".
expect(page.get_by_role("table").get_by_role("cell", name=re.compile(r"^OK\b"))).to_be_visible(timeout=LOAD)
expect(page.get_by_role("cell", name="3 ok")).to_be_visible()
# 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
page.get_by_role("link", name="#1").click()
expect(page.get_by_role("heading", name="Run #1")).to_be_visible(timeout=LOAD)
# Run detail is tabbed, and Rows lands first. A person's results live inside the row that
# built them, so open every row before reading them.
open_rows(page)
# Every user is a clickable tab in the run's nav; the selected one's rows show below it.
for username in ("sarah", "mike", "jess"):
expect(page.get_by_role("tab", name=re.compile(username, re.IGNORECASE))).to_be_visible()
def test_run_detail_shows_what_changed(self, page: Page, app: ShortlistApp, reset_fake_plex):
"""The diff is the audit trail (rule 10): everything Added on the first run, and on the
next run the staleness guard rotates fresh titles in — without ever shrinking a row."""
state = reset_fake_plex
first = build_real_rows(app)
# The engine must record a per-(row, library) breakdown — the whole run page is built from it.
run = app.api("GET", f"/api/runs/{first['id']}").json()
assert any(b["added"] for u in run["users"] for b in u["breakdown"]), "no per-library breakdown recorded"
page.goto(f"/runs/{first['id']}")
expect(page.get_by_role("heading", name=f"Run #{first['id']}")).to_be_visible(timeout=LOAD)
open_rows(page)
# First run: every pick is new, so the selected user's row shows a "+N new" badge and nothing
# removed. Picks render as a ranked list, and their titles are answerable from here.
expect(page.get_by_text(re.compile(r"\+\d+ new")).first).to_be_visible()
expect(page.get_by_text(re.compile(r"\d+ removed"))).to_have_count(0)
# Asserted against a title the API says was actually delivered, rather than a pattern the
# demo library's names happen to match — those are real film and show titles now, chosen so
# the docs screenshots look like a real library, and they will change again.
delivered = next(p["title"] for u in run["users"] for b in u["breakdown"] for p in b["picks"])
expect(page.locator("body")).to_contain_text(delivered)
def row_sizes() -> dict[str, int]:
sizes: dict[str, int] = {}
for collection in state.collections.values():
for label in collection.labels:
if label.lower().startswith("shortlist_"):
sizes[label.lower()] = sizes.get(label.lower(), 0) + len(collection.item_keys)
return sizes
before = row_sizes()
second = build_real_rows(app)
page.goto(f"/runs/{second['id']}")
expect(page.get_by_role("heading", name=f"Run #{second['id']}")).to_be_visible(timeout=LOAD)
open_rows(page)
# "Don't repeat the last 3 runs' picks" rotates the row; it must never leave it short.
# When fresh candidates run out, held-back titles backfill — a row shrinking (or being
# pruned) because the pool was thin for one night would be a regression, not variety.
assert row_sizes() == before, "a re-run must keep every row exactly as full as it was"
for user in app.api("GET", f"/api/runs/{second['id']}").json()["users"]:
diff = user["diff"]
assert len(diff.get("added", [])) == len(diff.get("removed", [])), (
f"{user['slug']}: a rotated-out title must be replaced, not simply dropped"
)
# The second run's page renders its users (the per-user nav), so its results are reachable.
expect(page.get_by_role("tab").first).to_be_visible(timeout=LOAD)
def test_a_users_picks_explain_themselves(self, page: Page, app: ShortlistApp):
"""Every pick carries its "Because you watched …" reason — the product's whole promise."""
build_real_rows(app)
sarah_id = _users_by_name(app)["sarah"]["id"]
run_id = app.api("GET", "/api/runs").json()[0]["id"]
sarah_picks = next(
u["picks"] for u in app.api("GET", f"/api/runs/{run_id}").json()["users"] if u["slug"] == "sarah"
)
page.goto(f"/users/{sarah_id}")
expect(page.get_by_role("heading", name="sarah")).to_be_visible(timeout=LOAD)
# Every poster carries its own reason under the title, so the first such caption is what
# shows the page has loaded.
reason_re = re.compile(r"Because you watched .+|Worth another watch|They asked for these")
expect(page.get_by_text(re.compile(r"^Because you watched .+")).first).to_be_visible(timeout=LOAD)
# Rows collapse to the first 5 picks (collapseAfter=5), so expand every row before counting:
# this test asserts EVERY pick carries a reason, not just the first few.
# Re-query each iteration: clicking one toggle re-renders the DOM and stales the locator list.
while (toggle := page.get_by_role("button", name=re.compile(r"Show all \d+")).first).is_visible():
toggle.click()
# Every pick is a list item, and every one of them names why it is there. sarah watches movies
# AND TV, so both libraries' rows show a reason.
picks = page.get_by_role("listitem").filter(has_text=re.compile(r"suggested by"))
expect(picks).to_have_count(len(sarah_picks))
expect(picks.filter(has_text=reason_re)).to_have_count(len(sarah_picks))
titles = {p["title"] for p in sarah_picks}
assert titles & {t for t, _ in DEMO_MOVIES} and titles & {t for t, _ in DEMO_SHOWS}, (
"sarah's row should mix both libraries — otherwise this test proves nothing about them"
)
# Ranked, and the rank the engine chose is the rank shown.
expect(page.get_by_text(sarah_picks[0]["title"], exact=True).first).to_be_visible()
def test_the_jess_gets_the_cold_start_row_and_says_so(self, page: Page, app: ShortlistApp):
"""No history -> the popular-titles fallback, labelled honestly rather than faked."""
build_real_rows(app)
jess_id = _users_by_name(app)["jess"]["id"]
page.goto(f"/users/{jess_id}")
expect(page.get_by_role("heading", name="jess")).to_be_visible(timeout=LOAD)
expect(page.get_by_text("New viewer").first).to_be_visible()
expect(page.get_by_text("Popular on this server").first).to_be_visible()