jacuzzi/tests/smoke_thumbnails.py

#!/usr/bin/env python3
"""Thumbnail smoke tests: junk URLs, and images that fail once.

Run against a locally running backend:

    backend/main.py &
    .venv/bin/python tests/smoke_thumbnails.py

The listing and the image hosts are both served by this script rather than by a
provider, because what's under test is what the client does with awkward data:

  * a `thumb` that isn't a URL at all -- sxyprn's "latest" listing sends items
    whose thumb is the bare string "https:". Resolved against the page that is
    *our own* address, so the card used to race our own HTML as if it were a
    picture, pin our origin to the proxy for the rest of the session, and ask
    /api/image to fetch "https:" (a 400, every time).

  * a thumbnail whose first request fails. One blip used to mean an empty box
    for as long as the card stayed mounted.
"""
import base64
import json
import sys
from playwright.sync_api import sync_playwright

BASE = "http://127.0.0.1:5000/"
SERVER = "https://hottubapp.io"
CHANNEL = "xvideos"
CDN = "https://cdn.example-thumbs.test"

# 1x1 transparent PNG.
PIXEL = base64.b64decode(
    "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAAC0lEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg==")

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([]));
}"""

STATE = """() => Array.from(document.querySelectorAll('#video-grid .video-card')).map((card) => {
    const img = card.querySelector('img');
    return {
        id: card.dataset.videoId,
        src: img ? img.getAttribute('src') || '' : '',
        loaded: img ? img.naturalWidth > 0 : 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 listing():
    """Twelve ordinary items, one with the junk thumb, one that fails once."""
    items = []
    for i in range(12):
        items.append({
            "id": f"test:{i}",
            "title": f"Video {i}",
            "url": f"{CDN}/watch/{i}",
            "channel": "test",
            "duration": 60 + i,
            "thumb": f"{CDN}/thumb/{i}.png",
            "tags": [],
        })
    items[3]["thumb"] = "https:"        # what sxyprn's "latest" actually sends
    items[7]["thumb"] = f"{CDN}/flaky.png"
    return {"items": items, "pageInfo": {"hasNextPage": False}}


def main():
    c = Checks()
    image_proxy_calls = []
    flaky_hits = {"direct": 0, "proxy": 0}

    with sync_playwright() as p:
        browser = p.chromium.launch(args=[
            "--no-sandbox", "--disable-dev-shm-usage", "--disable-gpu",
        ])
        page = browser.new_page(viewport={"width": 1400, "height": 1000})

        def serve_listing(route):
            route.fulfill(status=200, content_type="application/json",
                          body=json.dumps(listing()))

        def serve_proxy(route):
            # Only the thumbnails matter here; record what the client asked us
            # to fetch on its behalf, and hand back the picture.
            image_proxy_calls.append(route.request.url)
            if "flaky.png" in route.request.url:
                flaky_hits["proxy"] += 1
                # The flaky picture is refused on *both* routes the first time
                # round, which is what used to leave the card empty for good.
                if flaky_hits["proxy"] == 1:
                    route.fulfill(status=502, content_type="text/plain", body="nope")
                    return
            route.fulfill(status=200, content_type="image/png", body=PIXEL)

        def serve_cdn(route):
            if route.request.url.endswith("/flaky.png"):
                flaky_hits["direct"] += 1
                if flaky_hits["direct"] == 1:
                    route.abort("connectionfailed")
                    return
            route.fulfill(status=200, content_type="image/png", body=PIXEL)

        page.route("**/api/videos", serve_listing)
        page.route("**/api/image*", serve_proxy)
        page.route(f"{CDN}/**", serve_cdn)

        page.goto(BASE, wait_until="domcontentloaded")
        page.evaluate(SEED, [SERVER, CHANNEL])
        page.goto(BASE, wait_until="load")
        page.wait_for_selector(".video-card", timeout=60000)
        # Long enough for the host race (2.5s of patience) and the retry ladder
        # (~900ms for its second step) to have run their course.
        page.wait_for_timeout(8000)

        cards = page.evaluate(STATE)
        by_id = {card["id"]: card for card in cards}

        print("\na thumb that isn't a URL")
        junk = by_id.get("test:3")
        c.ok("the card is mounted", junk is not None)
        if junk:
            c.ok("it asks for nothing at all", junk["src"] == "",
                 f"src={junk['src']!r}")
        c.ok("and nothing is sent to the image proxy for it",
             not [u for u in image_proxy_calls if "https%3A&" in u or u.endswith("url=https%3A")],
             str([u for u in image_proxy_calls if "https%3A" in u][:2]))

        print("\nthe rest of the page is unaffected by it")
        others = [card for card in cards if card["id"] != "test:3"]
        c.ok("every other card shows its picture",
             all(card["loaded"] for card in others),
             str([card["id"] for card in others if not card["loaded"]]))
        # The junk URL used to resolve to our own origin, whose race then failed
        # and pinned it to the proxy -- for everything, for the whole session.
        c.ok("our own origin is not pinned to the proxy",
             page.evaluate("() => App.videos.thumbnailUrl(location.origin + '/x.png')")
             == page.evaluate("() => location.origin + '/x.png'"))

        print("\na thumbnail refused on both routes, once")
        flaky = by_id.get("test:7")
        c.ok("the card is mounted", flaky is not None)
        c.ok("both routes were refused once",
             flaky_hits["direct"] >= 1 and flaky_hits["proxy"] >= 1,
             f"direct={flaky_hits['direct']} proxy={flaky_hits['proxy']}")
        c.ok("and it was asked for again after that",
             flaky_hits["direct"] + flaky_hits["proxy"] > 2,
             f"direct={flaky_hits['direct']} proxy={flaky_hits['proxy']}")
        if flaky:
            c.ok("so the card ends up showing a picture", flaky["loaded"],
                 f"src={flaky['src']!r}")
        c.ok("and a dead thumbnail stops being asked for",
             flaky_hits["direct"] + flaky_hits["proxy"] <= 4,
             f"direct={flaky_hits['direct']} proxy={flaky_hits['proxy']}")

        browser.close()

    print(f"\n{'OK' if not c.failed else 'FAILED'}: {c.failed} check(s) failed")
    return 1 if c.failed else 0


sys.exit(main())