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'),
        };
    });
}"""


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 scroll_around(page, downs=8):
    """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(700)
    page.wait_for_timeout(1500)
    for _ in range(downs):
        page.evaluate("() => window.scrollBy(0, -window.innerHeight * 1.5)")
        page.wait_for_timeout(500)
    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 "")


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")

        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)

        stats = page.evaluate(
            "() => (App.virtualGrid.stats && App.virtualGrid.stats()) || null")
        if stats:
            total = (stats.get("built", 0) + stats.get("recycled", 0)) or 1
            print(f"\npool: {stats.get('recycled', 0)} recycled / {total} mounts "
                  f"({100 * stats.get('recycled', 0) // total}% hit rate), "
                  f"{stats.get('pooled', 0)} idle in pool")

        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())