Files
jacuzzi/tests/smoke_poster_capture.py
Simon 4f772773e1 Let the reader have the captured pictures taken again
A frame taken out of a video is kept for good: that is what makes a dead
favorite cheap to scroll past, and it is also the one thing that can go wrong
with no way out. A picture from the wrong moment, or one kept by a version of
this that would keep a frame with nothing in it, stays on the card for the life
of the item.

So Settings grows a Captured Thumbnails section: what is kept and how much room
it takes, and a button that throws the lot away. The cards on screen go back to
the start of the thumbnail ladder rather than waiting to be scrolled out of the
window and back, which is where a fresh capture comes from, and the session's
memory of which items there was no point asking about goes with the pictures --
otherwise "take them again" would quietly skip every item that failed once.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01MPZFnNdHbPGDTqQUNiE4ZN
2026-10-21 21:08:06 +00:00

309 lines
14 KiB
Python

#!/usr/bin/env python3
"""A favorite whose thumbnail has expired gets a frame out of the video instead.
Run against a locally running backend:
backend/main.py &
.venv/bin/python tests/smoke_poster_capture.py
Everything the client would reach for is a fixture, and every one of them is
dead on purpose: the listed picture 404s on both routes, the image proxy 404s,
and the source page has no replacement to offer. That is the state a favorite
saved months ago is actually in -- the signature on its CDN thumbnail expired --
and what the test asks is that the card still ends up with a picture of the
video, scaled down, kept, and not fetched again.
"""
import json
import os
import re
import sys
from playwright.sync_api import sync_playwright
# 1280x720, eight seconds. Wide on purpose: the capture has to scale down to
# card size, and a fixture no wider than the cache's own width would never show
# whether it does.
CLIP = os.path.join(os.path.dirname(__file__), "fixtures", "wide.mp4")
BASE = "http://127.0.0.1:5000/"
SERVER = "https://fixture.test"
CHANNEL = "clips"
# What App.posters scales a frame down to, and what it keeps the cache under.
POSTER_WIDTH = 360
CACHE_KEY = "posterCache"
STATUS = {
"online": True,
"channels": [{
"id": CHANNEL,
"name": "Fixture Clips",
"description": "A channel that only exists for this test.",
"options": [],
}],
}
# The favorite is the first item of the listing, saved with the thumbnail it
# had at the time -- the one that no longer loads.
SEED = """([server, channel]) => {
localStorage.setItem('config', JSON.stringify({ servers: [{ [server]: {} }] }));
localStorage.setItem('preferences', JSON.stringify({ [server]: { channelId: channel } }));
localStorage.removeItem('session');
localStorage.removeItem('posterCache');
Object.keys(localStorage).forEach((k) => {
if (k.indexOf('posterCache:') === 0) localStorage.removeItem(k);
});
localStorage.setItem('favorites', JSON.stringify([{
key: 'vid-0',
id: 'vid-0',
url: 'https://fixture.test/v0.mp4',
title: 'kept clip 0',
thumb: 'https://pictures.fixture.test/gone/0.jpg',
channel: channel,
uploader: '',
duration: 8,
isLive: false,
favoriteDate: new Date().toISOString()
}]));
}"""
# The picture on a card, and whether it is one: a broken <img> has no
# intrinsic size.
SHOT = """(id) => {
const card = document.querySelector(`.video-card[data-video-id="${id}"]`);
if (!card) return null;
const img = card.querySelector('.video-thumb img');
return {
src: (img.getAttribute('src') || '').slice(0, 24),
length: (img.getAttribute('src') || '').length,
width: img.naturalWidth,
height: img.naturalHeight
};
}"""
CACHE = """() => {
const index = JSON.parse(localStorage.getItem('posterCache') || '[]');
return index.map((entry) => ({
bytes: entry.n,
stored: (localStorage.getItem(`posterCache:${entry.k}`) || '').slice(0, 24),
length: (localStorage.getItem(`posterCache:${entry.k}`) || '').length
}));
}"""
PIXEL = """([id, x, y]) => new Promise((resolve) => {
const card = document.querySelector(`.video-card[data-video-id="${id}"]`);
const img = card ? card.querySelector('.video-thumb img') : null;
if (!img) { resolve(null); return; }
const probe = new Image();
probe.onload = () => {
const canvas = document.createElement('canvas');
canvas.width = probe.width;
canvas.height = probe.height;
const ctx = canvas.getContext('2d');
ctx.drawImage(probe, 0, 0);
const px = ctx.getImageData(x, y, 1, 1).data;
resolve({ r: px[0], g: px[1], b: px[2] });
};
probe.onerror = () => resolve(null);
probe.src = img.getAttribute('src');
})"""
CAPTURED = """(id) => {
const card = document.querySelector(`.video-card[data-video-id="${id}"]`);
if (!card) return false;
const img = card.querySelector('.video-thumb img');
return (img.getAttribute('src') || '').indexOf('data:image/jpeg') === 0;
}"""
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 items():
# The Referer is what gives an item a source page, which is the repair that
# comes before a capture. A favorite has none -- normalize() keeps no
# headers, since theirs expire -- so a saved item goes straight to the
# frame, while the same item in the listing walks the whole ladder.
return [{"id": f"vid-{i}", "title": f"clip {i}",
"url": f"https://fixture.test/v{i}.mp4",
"thumb": f"https://pictures.fixture.test/gone/{i}.jpg",
"http_headers": {"Referer": "https://page.fixture.test/"},
"channel": CHANNEL, "duration": 8, "isLive": False, "tags": []}
for i in range(3)]
def main():
c = Checks()
streamed = []
posters = []
with sync_playwright() as p:
browser = p.chromium.launch(args=[
"--no-sandbox", "--disable-dev-shm-usage", "--disable-gpu",
"--autoplay-policy=no-user-gesture-required",
"--renderer-process-limit=1", "--js-flags=--max-old-space-size=512",
])
page = browser.new_page(viewport={"width": 1180, "height": 920})
crashes = []
page.on("pageerror", lambda e: crashes.append(str(e)))
page.route("**/api/status", lambda route: route.fulfill(
status=200, content_type="application/json", body=json.dumps(STATUS)))
page.route("**/api/videos", lambda route: route.fulfill(
status=200, content_type="application/json",
body=json.dumps({"items": items(), "pageInfo": {"hasNextPage": False}})))
# Every way to the listed picture is gone: the provider, and our own
# image proxy standing in front of it.
page.route(re.compile(r"pictures\.fixture\.test"),
lambda route: route.fulfill(status=404, body=""))
page.route(re.compile(r"/api/image"),
lambda route: route.fulfill(status=404, body=""))
# And the source page has no fresher one to read off.
def poster(route):
posters.append(route.request.url)
route.fulfill(status=200, content_type="application/json",
body=json.dumps({"thumb": ""}))
page.route(re.compile(r"/api/poster"), poster)
# The video itself, which is the one thing that still works. Served
# the way the real proxy serves it -- byte ranges and all -- because a
# response a browser cannot seek in is one it will not capture a frame
# a quarter of the way into.
def stream(route):
streamed.append(route.request.url)
body = open(CLIP, "rb").read()
asked = route.request.headers.get("range") or ""
span = re.match(r"bytes=(\d+)-(\d*)", asked)
if span:
start = int(span.group(1))
end = int(span.group(2)) if span.group(2) else len(body) - 1
chunk = body[start:end + 1]
route.fulfill(status=206, body=chunk, headers={
"Content-Type": "video/mp4",
"Accept-Ranges": "bytes",
"Content-Range": f"bytes {start}-{end}/{len(body)}",
"Content-Length": str(len(chunk)),
})
return
route.fulfill(status=200, body=body, headers={
"Content-Type": "video/mp4",
"Accept-Ranges": "bytes",
"Content-Length": str(len(body)),
})
page.route(re.compile(r"/api/stream"), stream)
page.goto(BASE, wait_until="domcontentloaded")
page.evaluate(SEED, [SERVER, CHANNEL])
page.goto(BASE, wait_until="load")
page.wait_for_selector(".video-card", timeout=30000)
print("\nthe listed picture is gone, on every route there is")
c.ok("the module is there", page.evaluate("() => typeof App.posters.capture") == 'function')
c.ok("nothing is cached to begin with", page.evaluate(CACHE) == [], str(page.evaluate(CACHE)))
print("\nso the card takes a frame out of the video")
page.wait_for_function(CAPTURED, arg="vid-0", timeout=40000)
shot = page.evaluate(SHOT, "vid-0")
c.ok("the favorite's thumbnail is a captured frame",
shot and shot["src"].startswith("data:image/jpeg"), str(shot))
c.ok("and it decoded, so it is a picture and not an empty box",
shot["height"] > 0, str(shot))
c.ok("scaled down to card width, not kept at video size",
shot["width"] == POSTER_WIDTH, str(shot["width"]))
c.ok("the aspect ratio came through",
abs(shot["height"] - round(POSTER_WIDTH * 720 / 1280)) <= 1, str(shot))
c.ok("it was asked for at the quarter mark, from the video itself",
any("/api/stream" in url for url in streamed), str(streamed[:2]))
c.ok("and the source page was asked for a fresh picture before that",
any("/api/poster" in url for url in posters), str(posters[:2]))
print("\nthe frame is the one a quarter of the way in")
# The fixture is three bands of flat colour, so the frame says when it
# was taken: red for the first second and a half, orange from there to
# two and a half -- which is where a quarter of eight seconds falls --
# and navy for the rest. A capture that drew the first frame it decoded
# instead of seeking would come out red, and one that waited for the
# middle, navy.
frame = page.evaluate(PIXEL, ["vid-0", 180, 100])
c.ok("the colour is the one the clip shows two seconds in",
frame is not None and frame["r"] > 190 and 100 < frame["g"] < 180 and frame["b"] < 110,
str(frame))
c.ok("so it is neither the opening frame nor the middle",
frame is not None and frame["g"] > 80 and not (frame["b"] > frame["r"]), str(frame))
print("\nand keeps it")
cached = page.evaluate(CACHE)
c.ok("one picture is in the cache", len(cached) == 1, str(cached))
c.ok("stored as a JPEG data URL",
cached[0]["stored"].startswith("data:image/jpeg"), str(cached[0]["stored"]))
c.ok("small enough to keep a hundred of them",
0 < cached[0]["length"] < 60000, str(cached[0]["length"]))
c.ok("and the index says how big it is",
cached[0]["bytes"] == cached[0]["length"], str(cached[0]))
print("\nthe videos nobody kept are left alone")
others = [page.evaluate(SHOT, "vid-1"), page.evaluate(SHOT, "vid-2")]
c.ok("a card that is not a favorite captures nothing",
all(not (shot and shot["src"].startswith("data:")) for shot in others), str(others))
c.ok("so exactly one video was fetched for a picture",
len(streamed) == 1, str(streamed))
print("\nand the next session does not fetch it again")
streamed.clear()
posters.clear()
page.goto(BASE, wait_until="load")
page.wait_for_selector(".video-card", timeout=30000)
page.wait_for_timeout(1500)
shot = page.evaluate(SHOT, "vid-0")
c.ok("the kept picture is on the card straight away",
shot and shot["src"].startswith("data:image/jpeg"), str(shot))
c.ok("without going back to the video", streamed == [], str(streamed))
c.ok("or to the source page", posters == [], str(posters))
c.ok("and it is still the same one picture in the cache",
len(page.evaluate(CACHE)) == 1, str(page.evaluate(CACHE)))
print("\nthe favorites bar shows it too")
bar = page.evaluate("""() => {
const img = document.querySelector('.favorite-card img');
return img ? (img.getAttribute('src') || '').slice(0, 24) : null;
}""")
c.ok("the bar card is the captured frame as well",
bar is not None and bar.startswith("data:image/jpeg"), str(bar))
print("\nand the reader can have them taken again")
streamed.clear()
page.click(".settings-toggle")
c.ok("the setting says what is being kept",
"1 picture kept" in page.inner_text("#clear-posters-status"),
page.inner_text("#clear-posters-status"))
page.click("#clear-posters-btn")
c.ok("nothing is kept any more", page.evaluate(CACHE) == [], str(page.evaluate(CACHE)))
c.ok("and it says what it threw away",
"Threw away 1 picture" in page.inner_text("#clear-posters-status"),
page.inner_text("#clear-posters-status"))
page.wait_for_function(CAPTURED, arg="vid-0", timeout=40000)
shot = page.evaluate(SHOT, "vid-0")
c.ok("the card has a frame out of the video again",
shot and shot["src"].startswith("data:image/jpeg"), str(shot))
c.ok("which was fetched for it, this being the point of the setting",
any("/api/stream" in url for url in streamed), str(streamed[:2]))
c.ok("and kept again, as one picture",
len(page.evaluate(CACHE)) == 1, str(page.evaluate(CACHE)))
c.ok("nothing threw along the way", not crashes, str(crashes[:2]))
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())