Files
jacuzzi/tests/smoke_grid.py
Simon 7624ca559a Close three holes in the card release path
A second review pass over the pool. All three are the same shape: state
or a listener outliving the video it belonged to.

attachProxyFallback replaces whatever fallback an image currently has,
so a call arriving late -- a race hitting its patience timeout after the
card was recycled -- took away the live listener and left a dead one, and
the new video's thumbnail would then fail with nothing behind it. This
one was self-inflicted: the detach came in last round to stop the
listeners accumulating, and introduced the clobber. It is token-guarded
now, like every other path that can arrive late.

The favourite pop is cleared by animationend, which never fires on a card
release() has already detached -- detached elements run no animations. So
the class rode into the pool and replayed on the next video the card
showed. Favourite something and flick-scroll to see it.

And withOrigin compared a token that dataset reports as undefined for an
unstamped card, which matched every unstamped card instead of none --
failing open in the guard whose whole purpose is noticing that the grid
has recycled the card out from under the player.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QBDkEXP4htyXTCZUwMLphd
2026-09-09 15:57:52 +00:00

239 lines
10 KiB
Python

#!/usr/bin/env python3
"""Grid smoke tests.
Run against a locally running backend:
backend/main.py &
.venv/bin/python tests/smoke_grid.py
The load-bearing check here is `content bleed`: every mounted card must render
the video its own data-video-id names. Nothing enforced that before cards were
recycled, because a card was thrown away the moment it left the window; once
cards are reused, a release path that forgets to clear something shows one
video's title, thumbnail or heart on another video's card.
"""
import sys
from playwright.sync_api import sync_playwright
BASE = "http://127.0.0.1:5000/"
SERVER = "https://hottubapp.io"
CHANNEL = "xvideos"
SEED = """([server, channel]) => {
localStorage.setItem('config', JSON.stringify({ servers: [{ [server]: {} }] }));
localStorage.setItem('preferences', JSON.stringify({ [server]: { channelId: channel } }));
localStorage.removeItem('session');
localStorage.setItem('favorites', JSON.stringify([]));
}"""
# Everything a mounted card renders, next to what its own id says it should.
INSPECT = """() => {
const byId = new Map();
(App.state.loadedVideos || []).forEach((v) => byId.set(String(v.id), v));
return Array.from(document.querySelectorAll('#video-grid .video-card')).map((card) => {
const id = card.dataset.videoId;
const v = byId.get(String(id)) || null;
const img = card.querySelector('img');
const dur = card.querySelector('.video-duration');
const up = card.querySelector('.video-uploader');
const fav = card.querySelector('.favorite-btn');
return {
id: id,
known: !!v,
title_shown: (card.querySelector('.video-title-text') || {}).textContent || '',
title_expected: v ? (v.title || '') : null,
// A thumbnail is served either straight from the provider or via
// /api/image?url=<encoded>; compare on the provider URL either way.
src_shown: (() => {
const raw = img ? (img.getAttribute('src') || '') : '';
if (!raw) return '';
try {
const u = new URL(raw, location.href);
return u.pathname === '/api/image'
? (u.searchParams.get('url') || raw) : raw;
} catch (e) { return raw; }
})(),
thumb_expected: v ? (v.thumb || '') : null,
duration_shown: dur && !dur.hidden ? dur.textContent : '',
duration_expected: v ? (App.videos.formatDuration(v.duration) || '') : null,
uploader_shown: up && !up.hidden ? (up.dataset.uploader || up.textContent || '') : '',
uploader_expected: v ? (v.uploader || '') : null,
heart_shown: fav ? fav.classList.contains('is-favorite') : null,
heart_expected: v ? App.favorites.has(v) : null,
stale_loading: card.classList.contains('is-loading'),
stale_pop: fav ? fav.classList.contains('just-favorited') : false,
};
});
}"""
class Checks:
def __init__(self):
self.failed = 0
def ok(self, label, condition, detail=""):
mark = "PASS" if condition else "FAIL"
if not condition:
self.failed += 1
print(f" [{mark}] {label}" + (f" -- {detail}" if detail and not condition else ""))
def boot(page):
"""Seed a known server/channel, then wait for the grid to fill.
Startup renders from the status cached in localStorage and refreshes it in
the background, so the first visit has to wait for that round trip before
any video is loaded.
"""
page.goto(BASE, wait_until="domcontentloaded")
page.evaluate(SEED, [SERVER, CHANNEL])
page.goto(BASE, wait_until="load")
try:
page.wait_for_selector(".video-card", timeout=90000)
except Exception:
# One reload, in case the status refresh or the listing request failed.
page.goto(BASE, wait_until="load")
page.wait_for_selector(".video-card", timeout=90000)
page.wait_for_timeout(3000)
def grow(page, want=80, tries=14):
"""Load enough videos that the grid is taller than the mount window.
The virtualizer keeps everything within 1.2 viewports of the screen mounted,
so a short list never unmounts anything and never exercises recycling.
"""
for _ in range(tries):
if page.evaluate("() => App.state.loadedVideos.length") >= want:
break
page.evaluate("() => window.scrollTo(0, document.documentElement.scrollHeight)")
page.wait_for_timeout(2000)
return page.evaluate("() => App.state.loadedVideos.length")
def scroll_around(page, downs=10):
"""Churn the mount/unmount path: far down, then back to the top."""
for _ in range(downs):
page.evaluate("() => window.scrollBy(0, window.innerHeight * 1.5)")
page.wait_for_timeout(500)
page.wait_for_timeout(1200)
page.evaluate("() => window.scrollTo(0, 0)")
page.wait_for_timeout(1200)
for _ in range(downs // 2):
page.evaluate("() => window.scrollBy(0, window.innerHeight * 2.5)")
page.wait_for_timeout(400)
page.wait_for_timeout(1500)
def check_cards(c, cards, phase):
print(f"\n{phase}: {len(cards)} cards mounted")
c.ok(f"{phase}: cards are mounted", len(cards) > 0)
c.ok(f"{phase}: every card's id is a loaded video",
all(x["known"] for x in cards),
str([x["id"] for x in cards if not x["known"]][:3]))
ids = [x["id"] for x in cards]
c.ok(f"{phase}: no duplicate cards for one video", len(ids) == len(set(ids)))
for field in ("title", "duration", "uploader"):
bad = [x for x in cards
if x["known"] and (x[f"{field}_shown"] or "") != (x[f"{field}_expected"] or "")]
c.ok(f"{phase}: {field} matches the card's own video", not bad,
f"{len(bad)} mismatched, e.g. id={bad[0]['id']} "
f"shown={bad[0][f'{field}_shown']!r} expected={bad[0][f'{field}_expected']!r}"
if bad else "")
# The thumbnail may be served direct or through /api/image, so compare on
# the underlying provider URL rather than the literal src.
bad_src = [x for x in cards if x["known"] and x["src_shown"]
and x["thumb_expected"] and x["thumb_expected"] not in x["src_shown"]
and x["thumb_expected"].split("?")[0] not in x["src_shown"]]
c.ok(f"{phase}: thumbnail belongs to the card's own video", not bad_src,
f"{len(bad_src)} mismatched, e.g. id={bad_src[0]['id']}" if bad_src else "")
bad_heart = [x for x in cards if x["known"] and x["heart_shown"] != x["heart_expected"]]
c.ok(f"{phase}: heart state matches the card's own video", not bad_heart,
f"{len(bad_heart)} mismatched" if bad_heart else "")
stale = [x for x in cards if x["stale_loading"]]
c.ok(f"{phase}: no card left in the loading state", not stale,
f"{len(stale)} stuck" if stale else "")
# The favourite pop is animated away by animationend, which never fires on a
# card released mid-animation -- so it can ride into the pool and replay on
# whatever video the card is bound to next.
popping = [x for x in cards if x["stale_pop"]]
c.ok(f"{phase}: no card replaying the favourite animation", not popping,
f"{len(popping)} popping, e.g. id={popping[0]['id']}" if popping else "")
def main():
c = Checks()
with sync_playwright() as p:
browser = p.chromium.launch(args=["--no-sandbox"])
page = browser.new_page(viewport={"width": 1400, "height": 1000})
boot(page)
check_cards(c, page.evaluate(INSPECT), "on first render")
# Informational: how many videos it took to outgrow the mount window
# varies with viewport and page size. Whether that was *enough* is
# asserted properly at the end, on the pool's hit rate.
print(f"\ngrew the listing to {grow(page)} videos")
scroll_around(page)
check_cards(c, page.evaluate(INSPECT), "after scrolling down and back")
# Favouriting must land on the clicked card and survive remounting.
page.evaluate("""() => {
const card = document.querySelector('#video-grid .video-card');
card.querySelector('.favorite-btn').click();
}""")
page.wait_for_timeout(800)
favourited = page.evaluate("() => App.favorites.getAll().map(f => f.key)")
c.ok("favouriting stores exactly one entry", len(favourited) == 1, str(favourited))
scroll_around(page, downs=4)
cards = page.evaluate(INSPECT)
check_cards(c, cards, "after favouriting and scrolling")
# The menu still opens on a card that has been through the cycle.
opened = page.evaluate("""() => {
const card = document.querySelector('#video-grid .video-card');
card.querySelector('.video-menu-btn').click();
return card.querySelector('.video-menu').classList.contains('open');
}""")
c.ok("the card menu opens after recycling", opened)
# Tag clicks read the button's own text now, not a data attribute.
searched = page.evaluate("""() => {
const tag = document.querySelector('#video-grid .video-card .video-tag');
if (!tag) return 'no-tags';
const label = tag.textContent;
tag.click();
return document.getElementById('search-input').value === label ? 'ok' : 'mismatch';
}""")
c.ok("clicking a tag searches for it", searched in ("ok", "no-tags"), searched)
page.wait_for_timeout(1500)
stats = page.evaluate("""
() => (App.virtualGrid.stats && App.virtualGrid.stats()) || null
""")
if stats:
total = (stats.get("built", 0) + stats.get("recycled", 0)) or 1
rate = 100 * stats.get("recycled", 0) // total
print(f"\npool: {stats.get('recycled', 0)} recycled / {total} mounts "
f"({rate}% hit rate), {stats.get('pooled', 0)} idle in pool")
# A hit rate near zero means cards are still being built per mount,
# so none of the checks above actually exercised a recycled card.
c.ok("cards are actually being recycled", rate >= 50, f"{rate}% hit rate")
browser.close()
print(f"\n{'FAILED' if c.failed else 'OK'}: {c.failed} check(s) failed")
return 1 if c.failed else 0
if __name__ == "__main__":
sys.exit(main())