jacuzzi/tests/smoke_grid.py

#!/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:
        # Lean launch flags: this runs alongside the app's own server, and a
        # default Chromium spikes hard enough at startup to get itself killed on
        # a constrained box.
        browser = p.chromium.launch(args=[
            "--no-sandbox",
            "--disable-dev-shm-usage",
            "--disable-gpu",
            "--renderer-process-limit=1",
            "--js-flags=--max-old-space-size=512",
        ])
        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())