520 lines
22 KiB
Python
520 lines
22 KiB
Python
"""B1 FE walk — headless-chromium proof of UI-SPEC §11 on the real page.
|
|
|
|
Not part of `app.selfcheck` (needs playwright + a live server); run it after
|
|
starting the scratch server (the brief's isolation recipe):
|
|
|
|
.venv/bin/python -m app.fe_walk_b1 setup # optional — auto at start
|
|
|
|
env -u OPENROUTER_API_KEY .venv/bin/python -m app.serve \
|
|
--db /var/home/admin/dev/bike-tinder-b1/test-walk.sqlite3 \
|
|
--img-dir /var/home/admin/dev/bike-tinder-b1/test-img \
|
|
--host 127.0.0.1 --port 9454 # in another shell (mock forge)
|
|
|
|
.venv/bin/python -m app.fe_walk_b1 # walk (this file)
|
|
|
|
Setup is deterministic and idempotent: a fresh copy of the pristine B1 copy
|
|
DB (test-copy.sqlite3) each run, synthetic multi-photo listings (ids 701+)
|
|
with local image files staged into test-img from media-sample/, and one
|
|
swiped return listing (705) whose stored last_photo_idx drives §11.5
|
|
rotation. Everything walks the REAL page: real pointers, real DOM, no
|
|
stubbed APIs.
|
|
"""
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import sqlite3
|
|
import sys
|
|
import time
|
|
import urllib.request
|
|
|
|
ARENA = '/var/home/admin/dev/bike-tinder-b1'
|
|
PRISTINE = os.path.join(ARENA, 'test-copy.sqlite3')
|
|
WALK_DB = os.path.join(ARENA, 'test-walk.sqlite3')
|
|
IMG_DIR = os.path.join(ARENA, 'test-img')
|
|
MEDIA = os.path.join(ARENA, 'media-sample')
|
|
BASE_URL = 'http://127.0.0.1:9454'
|
|
|
|
CDN = ('https://media.kijiji.ca/api/v1/ca-prod-dealer-ads/images/x/%s'
|
|
'?rule=kijijica-640-webp')
|
|
|
|
# synthetic fixtures: listing_id -> (n_photos, label)
|
|
FIX = {
|
|
701: (8, 'gallery-8'),
|
|
702: (3, 'gallery-3'),
|
|
703: (2, 'gallery-2'),
|
|
704: (1, 'single-photo'),
|
|
705: (8, 'return-rotation'), # pre-swiped: last_photo_idx=0 -> lead 1
|
|
706: (0, 'zero-photo'),
|
|
}
|
|
SCORE_BASE = 2000
|
|
|
|
|
|
def stage_img_files():
|
|
"""{id}_0..{id}_{n-1}.webp (and {id}.webp for the legacy single) from
|
|
media-sample's real webp files — no network needed."""
|
|
os.makedirs(IMG_DIR, exist_ok=True)
|
|
names = sorted(m for m in os.listdir(MEDIA) if m.endswith('.webp'))
|
|
assert len(names) == 12, names
|
|
k = 0 # cycle through the 12 real files
|
|
for lid in sorted(FIX):
|
|
n = FIX[lid][0]
|
|
if n == 0:
|
|
continue
|
|
if n == 1:
|
|
shutil.copy2(os.path.join(MEDIA, names[k % len(names)]),
|
|
os.path.join(IMG_DIR, '%d.webp' % lid))
|
|
k += 1
|
|
continue
|
|
for i in range(n):
|
|
shutil.copy2(os.path.join(MEDIA, names[k % len(names)]),
|
|
os.path.join(IMG_DIR, '%d_%d.webp' % (lid, i)))
|
|
k += 1
|
|
|
|
|
|
def setup_db():
|
|
shutil.copy2(PRISTINE, WALK_DB)
|
|
from app import db as dbmod
|
|
conn = dbmod.connect(WALK_DB)
|
|
dbmod.init_db(conn) # B1 migrations
|
|
for lid in sorted(FIX):
|
|
n = FIX[lid][0]
|
|
imgs = [CDN % ('x%d-%d' % (lid, i)) for i in range(n)]
|
|
img = imgs[0] if imgs else None
|
|
score = SCORE_BASE + (750 - lid) # 701 highest -> top of the deck
|
|
conn.execute(
|
|
"DELETE FROM listings WHERE id=?", (lid,))
|
|
conn.execute(
|
|
"INSERT INTO listings (id, region, src_id, title, url, status,"
|
|
" img_url, imgs, score, first_seen, desc) VALUES"
|
|
" (?, 'ns', ?, ?, ?, 'active', ?, ?, ?,"
|
|
" '2026-11-11T00:00:00Z', ?)",
|
|
(lid, 'b1fix%d' % lid, 'B1 test %s' % FIX[lid][1],
|
|
'https://www.kijiji.ca/v-motorcycles/b1/%d' % lid,
|
|
img, json.dumps(imgs) if imgs else None, score, 'B1 fixture'))
|
|
conn.execute(
|
|
"INSERT OR REPLACE INTO swipes (listing_id, action, created_at,"
|
|
" last_photo_idx) VALUES (705, 'like', '2026-11-08T00:00:00Z', 0)")
|
|
# Close live-copied sections: the first /api/section read then composes
|
|
# a fresh 25 on the walk DB (deterministic fixture ordering).
|
|
conn.execute(
|
|
"UPDATE sections SET status='closed' WHERE status != 'closed'")
|
|
# A round-1 event per fixture keeps synthetics out of the wildcard
|
|
# pool (wildcards must be unseen); new-pool eligibility is unaffected.
|
|
# 705's event matches its stored verdict (round 1, like).
|
|
for lid in FIX:
|
|
conn.execute(
|
|
"INSERT INTO swipe_events (listing_id, action, round,"
|
|
" section_id) VALUES (?, ?, 1, NULL)",
|
|
(lid, 'like' if lid == 705 else 'maybe'))
|
|
conn.commit()
|
|
n = conn.execute('SELECT COUNT(*) AS n FROM listings').fetchone()['n']
|
|
conn.close()
|
|
print('setup: walk DB at %s (%s listings; 701-706 synthetic, 705 = liked'
|
|
' return w/ last_photo_idx=0)' % (WALK_DB, n))
|
|
|
|
|
|
def need_server():
|
|
try:
|
|
with urllib.request.urlopen(BASE_URL + '/api/health',
|
|
timeout=3) as r:
|
|
return r.status == 200
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
# --- walk helpers (module-level `page` assigned inside walk) ----------------
|
|
|
|
def wait(selector, timeout=8000):
|
|
page.wait_for_selector(selector, timeout=timeout)
|
|
|
|
|
|
def card_box():
|
|
b = page.evaluate("""() => { const el=document.querySelector('.topcard');
|
|
if(!el) return null; const r=el.getBoundingClientRect();
|
|
return {x:r.x,y:r.y,w:r.width,h:r.height}; }""")
|
|
assert b, 'no .topcard in DOM'
|
|
return b
|
|
|
|
|
|
def top_state():
|
|
return page.evaluate("""() => {
|
|
const t=document.querySelector('.topcard');
|
|
const r={topcards:document.querySelectorAll('.topcard').length,
|
|
hasSegs:!!t.classList.contains('hasSegs'), segs:0, segOn:-1,
|
|
photo:null, nophoto:false, band:null,
|
|
stampLike:0, stampNope:0, stampMaybe:0};
|
|
const s=t.querySelector('.segments');
|
|
if(s){ r.segs=s.children.length;
|
|
[...s.children].forEach((x,k)=>{ if(x.classList.contains('on'))
|
|
r.segOn=k; }); }
|
|
const p=t.querySelector('.phwrap img.photo');
|
|
if(p) r.photo=p.getAttribute('src');
|
|
else if(t.querySelector('.nophoto')) r.nophoto=true;
|
|
const b=t.querySelector('.band'); if(b) r.band=b.textContent;
|
|
const sl=t.querySelector('.stamp.like');
|
|
if(sl) r.stampLike=sl.style.opacity;
|
|
const sn=t.querySelector('.stamp.nope');
|
|
if(sn) r.stampNope=sn.style.opacity;
|
|
const sm=t.querySelector('.stamp.maybe');
|
|
if(sm) r.stampMaybe=sm.style.opacity;
|
|
return r; }""")
|
|
|
|
|
|
def tap_third(which):
|
|
b = card_box()
|
|
y = b['y'] + b['h'] * 0.72
|
|
x = b['x'] + b['w'] * (0.16 if which == 'left'
|
|
else (0.84 if which == 'right' else 0.5))
|
|
page.mouse.click(x, y)
|
|
time.sleep(0.08)
|
|
|
|
|
|
def assert_item(cond, msg):
|
|
if not cond:
|
|
print('FAIL ' + msg)
|
|
sys.exit(1)
|
|
print('ok ' + msg)
|
|
|
|
|
|
def lead_of(item):
|
|
n = item.get('n_photos') or 0
|
|
li = item.get('last_photo_idx')
|
|
if not n or n < 2 or li is None:
|
|
return 0
|
|
return (li + 1) % n
|
|
|
|
|
|
def photo_url(item, i):
|
|
if not item.get('img'):
|
|
return None
|
|
return item['img'] if i == 0 else '/img/%s?i=%s' % (item['id'], i)
|
|
|
|
|
|
def viewer_on():
|
|
return page.evaluate(
|
|
"() => document.getElementById('viewer').classList.contains('on')")
|
|
|
|
|
|
def vimg_src():
|
|
return page.evaluate(
|
|
"() => document.getElementById('vimg').getAttribute('src')")
|
|
|
|
|
|
def vimg_transform():
|
|
return page.evaluate(
|
|
"() => document.getElementById('vimg').style.transform")
|
|
|
|
|
|
# --- the walk ---------------------------------------------------------------
|
|
|
|
def walk():
|
|
from playwright.sync_api import sync_playwright
|
|
with sync_playwright() as pw:
|
|
exe = os.path.expanduser(
|
|
'~/.cache/ms-playwright/chromium-1243/chrome-linux64/chrome')
|
|
if not os.path.exists(exe):
|
|
exe = None
|
|
browser = pw.chromium.launch(headless=True, executable_path=exe,
|
|
args=['--no-sandbox'])
|
|
ctx = browser.new_context(viewport={'width': 390, 'height': 844},
|
|
has_touch=True, is_mobile=True)
|
|
global page
|
|
page = ctx.new_page()
|
|
page.goto(BASE_URL + '/', wait_until='networkidle')
|
|
wait('.topcard')
|
|
time.sleep(0.6)
|
|
|
|
section = None
|
|
for _ in range(40):
|
|
body = page.evaluate(
|
|
"() => fetch('/api/section').then(r=>r.json())")
|
|
if body and body.get('items'):
|
|
section = body
|
|
break
|
|
time.sleep(0.5)
|
|
assert section, 'no section payload'
|
|
items = list(section['items'])
|
|
print('section: %d items — recipe %s' % (len(items), json.dumps(
|
|
section['section']['recipe'], sort_keys=True)))
|
|
by_id = {i['id']: i for i in items}
|
|
for lid in (701, 702, 703, 704, 706):
|
|
assert lid in by_id, 'fixture %s missing from section' % lid
|
|
assert by_id[lid]['slot'] == 'new', (lid, by_id[lid]['slot'])
|
|
assert by_id[705]['slot'] == 'return', by_id[705]['slot']
|
|
|
|
seen = []
|
|
round_no = 0
|
|
while True:
|
|
remaining = [i['id'] for i in items if i['id'] not in seen]
|
|
if not remaining:
|
|
break
|
|
t = top_state()
|
|
top_id = None
|
|
m = re.search(r'/img/(\d+)(?:\?i=\d+)?$', t['photo'] or '')
|
|
if m:
|
|
top_id = int(m.group(1))
|
|
else:
|
|
for i in items: # zero-photo card: no src to match
|
|
if i['id'] not in seen and not i.get('img'):
|
|
top_id = i['id']
|
|
break
|
|
assert top_id in by_id, 'cannot identify top card'
|
|
item = by_id[top_id]
|
|
seen.append(top_id)
|
|
round_no += 1
|
|
n = item.get('n_photos') or 0
|
|
exp_lead = lead_of(item)
|
|
exp_src = photo_url(item, exp_lead)
|
|
print('-- top id=%s [%s] n=%d lead=%d slot=%s --'
|
|
% (top_id, item['title'][:26], n, exp_lead, item['slot']))
|
|
assert_item(t['topcards'] == 1,
|
|
'exactly one .topcard before gestures')
|
|
if n >= 2:
|
|
assert_item(t['hasSegs'] and t['segs'] == n,
|
|
'segments count == n (%d)' % n)
|
|
assert_item(t['segOn'] == exp_lead,
|
|
'segment idx == rotation lead (%d)' % exp_lead)
|
|
assert_item(t['photo'] == exp_src,
|
|
'photo src == expected lead url')
|
|
else:
|
|
assert_item(not t['hasSegs'] and t['segs'] == 0,
|
|
'no segments for n=%d' % n)
|
|
if n == 1:
|
|
assert_item(t['photo'] == photo_url(item, 0),
|
|
'single photo shows the primary')
|
|
else:
|
|
assert_item(t['nophoto'],
|
|
'zero-photo card shows the placeholder')
|
|
if item['slot'] == 'return' and item['id'] == 705:
|
|
assert_item(exp_lead == 1, 'rotation lead: (0+1)%%8 = 1')
|
|
assert_item(t['band'] and 'SECOND LOOK' in t['band'],
|
|
'return card carries the SECOND LOOK band '
|
|
'(with segments)')
|
|
|
|
if n >= 2:
|
|
# --- flip: advance + clamp both ends (no wrap) ----------
|
|
for i in range(n + 2):
|
|
tap_third('right')
|
|
st = top_state()
|
|
assert_item(st['topcards'] == 1,
|
|
'tap kept the one-.topcard invariant')
|
|
want = min(exp_lead + i + 1, n - 1)
|
|
assert_item(st['segOn'] == want,
|
|
'right-tap %d -> idx %d (clamp)' % (i + 1,
|
|
want))
|
|
for i in range(n + 2):
|
|
tap_third('left')
|
|
st = top_state()
|
|
assert_item(st['topcards'] == 1,
|
|
'tap kept the one-.topcard invariant')
|
|
want = max(n - 1 - (i + 1), 0)
|
|
assert_item(st['segOn'] == want,
|
|
'left-tap %d -> idx %d (clamp)' % (i + 1,
|
|
want))
|
|
st = top_state()
|
|
assert_item(st['topcards'] == 1
|
|
and st['stampLike'] in ('0', '')
|
|
and st['stampNope'] in ('0', '')
|
|
and st['stampMaybe'] in ('0', ''),
|
|
'taps never commit: no stamps, no flight')
|
|
# --- viewer: open on current photo, nav + clamp, zoom,
|
|
# pull-close restores the same card+photo ------------
|
|
want_idx = 1
|
|
while st['segOn'] > want_idx:
|
|
tap_third('left')
|
|
st = top_state()
|
|
if st['segOn'] < want_idx:
|
|
tap_third('right')
|
|
st = top_state()
|
|
before_src = st['photo']
|
|
tap_third('center')
|
|
assert_item(viewer_on(), 'center tap opens the viewer')
|
|
vsegs = page.evaluate(
|
|
"() => document.getElementById('vsegs')"
|
|
".querySelectorAll('i').length")
|
|
assert_item(vsegs == n, 'viewer segments == n (%d)' % n)
|
|
assert_item(vimg_src() == before_src,
|
|
'viewer opens on the current photo')
|
|
vp = page.locator('#vphoto')
|
|
vb = vp.bounding_box()
|
|
al, ar = (vb['x'] + vb['width'] * 0.75,
|
|
vb['x'] + vb['width'] * 0.15)
|
|
ay = vb['y'] + vb['height'] * 0.5
|
|
page.mouse.move(al, ay)
|
|
page.mouse.down()
|
|
page.mouse.move(ar, ay, steps=6)
|
|
page.mouse.up()
|
|
time.sleep(0.1)
|
|
land = min(want_idx + 1, n - 1)
|
|
assert_item(vimg_src() == photo_url(item, land),
|
|
'viewer swipe-left advances + clamps')
|
|
page.mouse.move(ar, ay)
|
|
page.mouse.down()
|
|
page.mouse.move(al, ay, steps=6)
|
|
page.mouse.up()
|
|
time.sleep(0.1)
|
|
back = max(land - 1, 0)
|
|
assert_item(vimg_src() == photo_url(item, back),
|
|
'viewer swipe-right steps back (clamped)')
|
|
page.mouse.dblclick(vb['x'] + vb['width'] * 0.5,
|
|
vb['y'] + vb['height'] * 0.5)
|
|
time.sleep(0.1)
|
|
assert_item('scale(2.5)' in vimg_transform(),
|
|
'double-tap zooms to 2.5x')
|
|
page.mouse.dblclick(vb['x'] + vb['width'] * 0.5,
|
|
vb['y'] + vb['height'] * 0.5)
|
|
time.sleep(0.1)
|
|
assert_item('scale(2.5)' not in vimg_transform(),
|
|
'double-tap again returns to 1x')
|
|
# pinch via synthetic pointer events (real handler path)
|
|
page.evaluate("""() => {
|
|
const v=document.getElementById('vphoto');
|
|
const r=v.getBoundingClientRect();
|
|
const mk=(t,id,x,y)=>{ v.dispatchEvent(new PointerEvent(t,
|
|
{pointerId:id, clientX:x, clientY:y, bubbles:true})); };
|
|
const x=r.width/2, y=r.height/2;
|
|
mk('pointerdown',10,x-25,y); mk('pointerdown',11,x+25,y);
|
|
for(let s=0;s<6;s++){ mk('pointermove',10,x-25-s*22,y);
|
|
mk('pointermove',11,x+25+s*22,y); }
|
|
mk('pointerup',10,x-25-6*22,y);
|
|
mk('pointerup',11,x+25+6*22,y);
|
|
}""")
|
|
assert_item('scale(4)' in vimg_transform(),
|
|
'pinch-zoom clamps at 4x')
|
|
page.mouse.dblclick(vb['x'] + vb['width'] * 0.5,
|
|
vb['y'] + vb['height'] * 0.5)
|
|
time.sleep(0.1) # back to 1x for the pull gesture
|
|
page.mouse.move(vb['x'] + vb['width'] * 0.5,
|
|
vb['y'] + vb['height'] * 0.35)
|
|
page.mouse.down()
|
|
page.mouse.move(vb['x'] + vb['width'] * 0.5,
|
|
vb['y'] + vb['height'] * 0.35 + 170, steps=8)
|
|
page.mouse.up()
|
|
time.sleep(0.2)
|
|
assert_item(not viewer_on(),
|
|
'pull-down (>110px) closes the viewer')
|
|
st = top_state()
|
|
assert_item(st['segOn'] == want_idx
|
|
and st['photo'] == before_src,
|
|
'close returns the same card + photo position')
|
|
else:
|
|
if n == 1:
|
|
tap_third('center')
|
|
assert_item(viewer_on(),
|
|
'center tap opens the viewer (single photo)')
|
|
page.locator('#vclose').click()
|
|
time.sleep(0.15)
|
|
if n == 0:
|
|
tap_third('center')
|
|
assert_item(not viewer_on(), 'zero-photo: no viewer')
|
|
|
|
# --- ⓘ beats the zones --------------------------------------
|
|
page.locator('.topcard .ibtn').click()
|
|
time.sleep(0.15)
|
|
assert_item(page.evaluate(
|
|
"() => document.getElementById('detail')"
|
|
".classList.contains('on')"),
|
|
'ⓘ opens the dossier (zones do not fire)')
|
|
pre = top_state()
|
|
page.locator('#dclose').click()
|
|
time.sleep(0.2)
|
|
post = top_state()
|
|
assert_item(pre['photo'] == post['photo']
|
|
and pre['segOn'] == post['segOn'],
|
|
'dossier close leaves the photo position intact')
|
|
if n >= 1:
|
|
# dossier hero -> viewer at photo 0 (dossier stays open)
|
|
page.locator('.topcard .ibtn').click()
|
|
time.sleep(0.15)
|
|
hb = page.locator('.dhero').bounding_box()
|
|
page.mouse.click(hb['x'] + hb['width'] / 2,
|
|
hb['y'] + hb['height'] / 2)
|
|
time.sleep(0.15)
|
|
assert_item(viewer_on() and page.evaluate(
|
|
"() => document.getElementById('detail')"
|
|
".classList.contains('on')"),
|
|
'dossier hero tap opens the viewer over the dossier')
|
|
assert_item(vimg_src() == photo_url(item, 0),
|
|
'dossier viewer starts at photo 0')
|
|
page.locator('#vclose').click()
|
|
time.sleep(0.15)
|
|
assert_item(not viewer_on() and page.evaluate(
|
|
"() => document.getElementById('detail')"
|
|
".classList.contains('on')"),
|
|
'viewer ✕ returns to the still-open dossier')
|
|
page.locator('#dclose').click()
|
|
time.sleep(0.15)
|
|
|
|
# single commit per round (fast chain runs after the report)
|
|
page.locator('#btnNope').click()
|
|
time.sleep(0.07)
|
|
|
|
# --- P3 regression: report -> undo x3 -> fast chain -> forge -----
|
|
wait('#report.on', 15000)
|
|
print('report visible: %s' % page.locator('#repCounts').inner_text())
|
|
assert_item('Section complete' in page.locator('#report').inner_text(),
|
|
'report screen after the full section')
|
|
for _ in range(4):
|
|
page.locator('#btnUndo').click()
|
|
time.sleep(0.08)
|
|
assert_item(page.evaluate(
|
|
"() => document.querySelectorAll('.topcard').length") == 1,
|
|
'undo x4 from the report re-opens the deck (one .topcard)')
|
|
for k in range(3):
|
|
b = card_box()
|
|
y = b['y'] + b['h'] * 0.6
|
|
page.mouse.move(b['x'] + b['w'] * 0.5, y)
|
|
page.mouse.down()
|
|
page.mouse.move(b['x'] + b['w'] * 0.5 + 320, y, steps=5)
|
|
page.mouse.up()
|
|
if k < 2:
|
|
time.sleep(0.07)
|
|
tc = page.evaluate(
|
|
"() => document.querySelectorAll('.topcard').length")
|
|
assert tc == 1, 'fast chain: topcard=%s (must be 1)' % tc
|
|
assert_item(True, '3 like-commits @70ms: one .topcard, no lockout')
|
|
page.locator('#btnNope').click() # finish the 4th restored card
|
|
time.sleep(0.1)
|
|
wait('#report.on', 15000)
|
|
page.locator('#btnForge').click()
|
|
wait('.topcard', 15000)
|
|
time.sleep(0.4)
|
|
sect = page.evaluate(
|
|
"() => document.getElementById('sect').textContent.trim()")
|
|
assert_item(re.match(r'^\d+/25$', sect) is not None,
|
|
'forge hydrates a fresh section (progress=%s)' % sect)
|
|
# --- List + Trends tabs render with P3 elements intact ----------
|
|
page.locator('.tab[data-tab="list"]').click()
|
|
wait('.row')
|
|
tanks = page.locator('.row .tank').count()
|
|
rows = page.locator('.row').count()
|
|
assert_item(tanks > 0 and rows > 0,
|
|
'list: %d rows, %d tank gauges (P3 intact)'
|
|
% (rows, tanks))
|
|
page.locator('.tab[data-tab="trends"]').click()
|
|
wait('#tLead')
|
|
assert_item((page.locator('#tLead').text_content() or '').strip()
|
|
!= '', 'trends tab renders')
|
|
print('WALK COMPLETE — all B1 FE checks green')
|
|
browser.close()
|
|
|
|
|
|
def main(argv=None):
|
|
if 'setup' in sys.argv or not os.path.exists(WALK_DB):
|
|
if not os.path.exists(PRISTINE):
|
|
print('pristine copy DB missing: %s' % PRISTINE)
|
|
return 1
|
|
setup_db()
|
|
stage_img_files()
|
|
print('fixtures staged in %s' % IMG_DIR)
|
|
if not need_server():
|
|
print('NO SERVER on %s — start it (see module docstring).' % BASE_URL)
|
|
return 1
|
|
walk()
|
|
return 0
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|