A video

#!/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. * a thumbnail that is simply gone. sxyprn's CDN paths are signed with an expiry and the listing hands out URLs that have passed theirs, so the newest posts arrive pointing at a 404. The item's own page still shows a working picture, and /api/poster reads it off. """ import base64 import json import os import sys import threading import urllib.error import urllib.parse import urllib.request from http.server import BaseHTTPRequestHandler, HTTPServer from playwright.sync_api import sync_playwright BASE = os.environ.get("JACUZZI_BASE", "http://127.0.0.1:5000/") SERVER = "https://hottubapp.io" CHANNEL = "xvideos" CDN = "https://cdn.example-thumbs.test" # Where a "provider page" lives, and the picture it says it has -- the shape # /api/poster reads. The proxy path mirrors the Hot Tub server's own, so the # client has to rebuild the site address from the item's Referer to find it. SITE = "https://site.example-thumbs.test" PROXY_PAGE = "https://proxy.example-thumbs.test/proxy/test/post/gone.html" # 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, }; })""" PAGE_HTML = """

""" class PageServer(threading.Thread): """A page for /api/poster to read, and one that redirects to a video -- which is what the Hot Tub proxy does, and what must not be downloaded.""" def __init__(self): super().__init__(daemon=True) outer = self class Handler(BaseHTTPRequestHandler): def log_message(self, *args): pass def do_GET(self): if self.path.startswith("/post"): body = PAGE_HTML.encode() self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) return if self.path.startswith("/video"): body = b"\0" * (4 * 1024 * 1024) self.send_response(200) self.send_header("Content-Type", "video/mp4") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) return self.send_response(404) self.end_headers() self.httpd = HTTPServer(("127.0.0.1", 0), Handler) outer.port = self.httpd.server_port def run(self): self.httpd.serve_forever() def stop(self): self.httpd.shutdown() GONE_STATE = """() => { const card = document.querySelector('#video-grid .video-card[data-video-id="test:9"]'); const img = card && card.querySelector('img'); return { src: img ? img.getAttribute('src') || '' : '', loaded: !!(img && img.naturalWidth > 0) }; }""" def gone_state(page): return page.evaluate(GONE_STATE) 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" # The expired-signature case: a dead thumbnail, and an item that says # enough about where it came from for the page to be found. items[9]["thumb"] = f"{CDN}/expired/gone.jpg" items[9]["url"] = PROXY_PAGE items[9]["channel"] = "test" items[9]["http_headers"] = {"Referer": f"{SITE}/"} return {"items": items, "pageInfo": {"hasNextPage": False}} def poster_of(page_url): """What /api/poster answers for a page, straight from the backend. Returns None if the backend doesn't have the endpoint -- which on a machine that has been running since before it existed means "restart it", not "broken".""" url = BASE.rstrip("/") + "/api/poster?url=" + urllib.parse.quote(page_url, safe="") try: with urllib.request.urlopen(url, timeout=30) as response: return json.load(response) except urllib.error.HTTPError as err: if err.code == 404: return None raise def main(): c = Checks() image_proxy_calls = [] poster_asks = [] flaky_hits = {"direct": 0, "proxy": 0} pages = PageServer() pages.start() local = f"http://127.0.0.1:{pages.port}" with sync_playwright() as p: browser = p.chromium.launch(args=[ "--no-sandbox", "--disable-dev-shm-usage", "--disable-gpu", # Lean flags, matching smoke_grid: 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. "--renderer-process-limit=1", "--js-flags=--max-old-space-size=512", ]) 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 "%2Fexpired%2F" in route.request.url or "/expired/" in route.request.url: route.fulfill(status=404, content_type="text/html", body="gone") return 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_poster(route): # The endpoint itself is exercised against a real page below; here # only the client's half is under test, so the answer is canned. asked = urllib.parse.parse_qs( urllib.parse.urlparse(route.request.url).query).get("url", [""])[0] poster_asks.append(asked) route.fulfill(status=200, content_type="application/json", body=json.dumps({"thumb": f"{CDN}/repaired.png"})) def serve_cdn(route): if "/expired/" in route.request.url: route.fulfill(status=404, content_type="text/html", body="gone") return 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/poster*", serve_poster) 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") # test:9's thumbnail is deliberately dead; it has a section of its own. others = [card for card in cards if card["id"] not in ("test:3", "test:9")] 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']}") print("\na thumbnail that is simply gone") gone = by_id.get("test:9") c.ok("the card is mounted", gone is not None) c.ok("its page was asked what picture it shows", poster_asks, str(poster_asks)) # The item points at a proxy path; the page lives on the site named by # the Referer the item carries. c.ok("and the page asked for was the item's own, on its own site", poster_asks and poster_asks[0] == f"{SITE}/post/gone.html", str(poster_asks[:2])) c.ok("the card ends up showing the replacement", gone and gone_state(page)["loaded"], str(gone_state(page))) c.ok("one ask is enough for that page", len(poster_asks) == 1, str(poster_asks)) print("\nwhat /api/poster reads off a page") answer = poster_of(f"{local}/post.html") if answer is None: c.ok("the backend has /api/poster (restart it if this fails)", False) answer = {} else: c.ok("the backend has /api/poster", True) c.ok("the page's own og:image, made absolute", answer.get("thumb") == f"http://pictures.example.test/fresh/poster.webp", str(answer)) video = poster_of(f"{local}/video.mp4") c.ok("a URL that turns out to be a video yields nothing", video.get("thumb") is None, str(video)) missing = poster_of(f"{local}/nope.html") c.ok("and so does a page that isn't there", missing.get("thumb") is None, str(missing)) browser.close() pages.stop() 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())