"""Dependency-free self-check for the Bike-Tinder data spine. python3 -m app.selfcheck Covers: gate rejections, dedupe (incl. false-positive sanity), scoring spot-checks against the reference rubric, gone-after-2-misses, the P2/P3 API surface (on temp copies of the DB only — the live file is never opened for writing), and the B1 photo-experience checks (gallery import, /img?i= chain, photo_idx persistence, rotation math, 640-row stability). """ import datetime import json import os import shutil import sys import tempfile from . import db, rubric from . import importer as imp def recent_iso(): return datetime.datetime.now(datetime.timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ') def make_item(title, price='$5,000', price_num=5000, km='10000', loc='Halifax', url='https://www.kijiji.ca/v-x/1', desc='', img=''): return {'title': title, 'price': price, 'price_num': price_num, 'loc': loc, 'km': km, 'url': url, 'desc': desc, 'img': img, 'first_seen': '2026-01-01T00:00:00Z', 'last_seen': '2026-01-01T00:00:00Z'} def make_state(items, ok=True, partial=False, fetched_at=None, absent=None): return {'source': 'kijiji-ns', 'fetched_at': fetched_at or recent_iso(), 'ok': ok, 'fail_streak': 0, 'pages': {'1': len(items)}, 'partial': partial, 'items': items, 'absent': absent or {}} def write_states(state_dir, ns, nb): os.makedirs(state_dir, exist_ok=True) for name, st in (('kijiji-ns', ns), ('kijiji-nb', nb)): with open(os.path.join(state_dir, name + '.json'), 'w', encoding='utf-8') as fh: json.dump(st, fh) def fresh_db(): d = tempfile.mkdtemp(prefix='bt-selfcheck-') return d, os.path.join(d, 'test.sqlite3') # --- gate rejections ------------------------------------------------------- def test_gate_rejections(): ok, err = imp.gate_region(make_state({'1': make_item('X')}, ok=False)) assert not ok and 'ok' in (err or ''), err ok, err = imp.gate_region(make_state({'1': make_item('X')}, partial=True)) assert not ok and 'partial' in (err or ''), err stale = (datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(hours=30)).strftime('%Y-%m-%dT%H:%M:%SZ') ok, err = imp.gate_region(make_state({'1': make_item('X')}, fetched_at=stale)) assert not ok and '26h' in (err or ''), err ok, err = imp.gate_region(make_state({})) assert not ok and 'empty' in (err or ''), err ok, err = imp.gate_region(make_state({'1': make_item('X')})) assert ok and err is None, err def test_gate_abort_leaves_db(): d, db_path = fresh_db() try: state_dir = os.path.join(d, 'state') write_states(state_dir, make_state({'1': make_item('X')}, ok=False), # ns fails make_state({'2': make_item('Y')})) # nb ok res = imp.run_import(state_dir, db_path) assert res['ok'] is False conn = db.connect(db_path) db.init_db(conn) n = conn.execute('SELECT COUNT(*) AS n FROM listings').fetchone()['n'] runs = conn.execute('SELECT ok, error FROM import_runs').fetchall() conn.close() assert n == 0, 'listings must be untouched on gate failure' assert len(runs) == 1 and runs[0]['ok'] == 0 and 'ns' in runs[0]['error'] finally: shutil.rmtree(d, ignore_errors=True) # --- dedupe ---------------------------------------------------------------- def test_dedupe_twins(): d, db_path = fresh_db() try: state_dir = os.path.join(d, 'state') twin = make_item('2021 Indian Scout Bobber', price_num=6500, km='8000') write_states(state_dir, make_state({'ns1': twin}), make_state({'nb1': dict(twin)})) res = imp.run_import(state_dir, db_path) assert res['ok'] and res['total'] == 2 and res['dup'] == 1, res conn = db.connect(db_path) db.init_db(conn) rows = conn.execute( 'SELECT id, region, status, dup_of FROM listings ORDER BY region').fetchall() conn.close() by_status = {r['status'] for r in rows} assert by_status == {'active', 'dup'}, rows dup = [r for r in rows if r['status'] == 'dup'][0] active = [r for r in rows if r['status'] == 'active'][0] assert dup['dup_of'] == active['id'], (dup, active) finally: shutil.rmtree(d, ignore_errors=True) def test_dedupe_false_positive_same_province(): # Same title+price but both in ONE province -> different bikes, not merged. d, db_path = fresh_db() try: state_dir = os.path.join(d, 'state') write_states(state_dir, make_state({'ns1': make_item('2025 BMW G 310 R', price_num=5444), 'ns2': make_item('2025 BMW G 310 R', price_num=5444)}), make_state({'nb_dummy': make_item('Some Other Bike', price_num=9999)})) res = imp.run_import(state_dir, db_path) assert res['ok'] and res['total'] == 3 and res['dup'] == 0, res finally: shutil.rmtree(d, ignore_errors=True) def test_dedupe_false_positive_price_diff(): # Same title, different price, cross-province -> not merged. d, db_path = fresh_db() try: state_dir = os.path.join(d, 'state') write_states(state_dir, make_state({'ns1': make_item('2021 Indian Scout', price_num=6500)}), make_state({'nb1': make_item('2021 Indian Scout', price_num=7000)})) res = imp.run_import(state_dir, db_path) assert res['ok'] and res['total'] == 2 and res['dup'] == 0, res finally: shutil.rmtree(d, ignore_errors=True) # --- scoring spot-checks --------------------------------------------------- def test_scoring(): s, fam = rubric.score_item('2021 Indian Scout Bobber', 'MVI done, low km', 6500, 8000, 2021) assert s == 105 and fam and 'Scout' in fam, (s, fam) s, fam = rubric.score_item('Kawasaki Vulcan 900 Custom', 'whitewalls, windshield', 5000, 20000, 2018) assert s == 102 and 'Vulcan 900 Custom' in fam, (s, fam) s, fam = rubric.score_item('Kawasaki Ninja 400', 'sport bike', 5000, 10000, 2020) assert s == 0 and fam is None, (s, fam) # vetoed -> unscored s, fam = rubric.score_item('Some random bike', '', 3000, None, None) assert s == 0 and fam is None, (s, fam) # no family -> unscored # km=0 placeholder -> treated as no km s, _ = rubric.score_item('2021 Indian Scout', '', 5000, 0, 2021) assert s == 92 + 5 + 2, s # base + price(<=5000) + year(>=2020), no km bonus # --- gone after 2 misses --------------------------------------------------- def test_gone_after_two_misses(): d, db_path = fresh_db() try: conn = db.connect(db_path) db.init_db(conn) conn.execute( "INSERT INTO listings (region, src_id, title, url, status, last_seen) " "VALUES ('ns','gone1','X','http://x/1','active','2026-01-01T00:00:01Z')") conn.commit() # single miss: last_seen == prev_good -> not gone gone = imp.mark_gone(conn, set(), '2026-01-01T00:00:01Z') assert gone == 0, gone st = conn.execute("SELECT status FROM listings WHERE src_id='gone1'").fetchone()['status'] assert st == 'active', st # second miss: last_seen < prev_good -> gone gone = imp.mark_gone(conn, set(), '2026-01-01T00:00:02Z') assert gone == 1, gone st = conn.execute("SELECT status FROM listings WHERE src_id='gone1'").fetchone()['status'] assert st == 'gone', st conn.close() finally: shutil.rmtree(d, ignore_errors=True) def test_gone_full_flow(): # End-to-end: present -> absent (single miss) -> absent (gone). d, db_path = fresh_db() try: state_dir = os.path.join(d, 'state') orig = imp.now_iso clock = {'now': '2026-01-01T00:00:01Z'} imp.now_iso = lambda: clock['now'] try: write_states(state_dir, make_state({'a1': make_item('2021 Indian Scout', price_num=6500)}), make_state({'nb_dummy': make_item('Some Other Bike', price_num=9999)})) imp.run_import(state_dir, db_path) # import 1: present clock['now'] = '2026-01-01T00:00:02Z' write_states(state_dir, make_state({'ns_other': make_item('Another Bike', price_num=8888)}), make_state({'nb_dummy': make_item('Some Other Bike', price_num=9999)})) imp.run_import(state_dir, db_path) # import 2: absent (single miss) conn = db.connect(db_path) db.init_db(conn) st = conn.execute("SELECT status FROM listings WHERE src_id='a1'").fetchone()['status'] conn.close() assert st == 'active', 'single miss must not mark gone' clock['now'] = '2026-01-01T00:00:03Z' imp.run_import(state_dir, db_path) # import 3: absent (2nd miss) conn = db.connect(db_path) db.init_db(conn) st = conn.execute("SELECT status FROM listings WHERE src_id='a1'").fetchone()['status'] conn.close() assert st == 'gone', '2nd consecutive miss must mark gone' finally: imp.now_iso = orig finally: shutil.rmtree(d, ignore_errors=True) def test_dedupe_multiunit_pairing(): # Cross-province group with 2 ns + 1 nb: min(ns,nb) = 1 pair -> 1 dup; # the surplus ns sibling stays active (no false merge of distinct bikes). d, db_path = fresh_db() try: state_dir = os.path.join(d, 'state') twin = '2025 BMW G 310 R' write_states(state_dir, make_state({'ns1': make_item(twin, price_num=5444, img='http://x/1.jpg'), 'ns2': make_item(twin, price_num=5444)}), make_state({'nb1': make_item(twin, price_num=5444)})) res = imp.run_import(state_dir, db_path) assert res['ok'] and res['total'] == 3 and res['dup'] == 1, res conn = db.connect(db_path) db.init_db(conn) nb = conn.execute( "SELECT status, dup_of FROM listings WHERE region='nb'").fetchone() assert nb['status'] == 'dup', nb canon = conn.execute( 'SELECT region, status, img_url FROM listings WHERE id=?', (nb['dup_of'],)).fetchone() assert canon['region'] == 'ns' and canon['status'] == 'active' assert canon['img_url'] is not None # best-complete ns was canonical # the surplus ns sibling stays active active_ns = conn.execute( "SELECT COUNT(*) AS n FROM listings WHERE region='ns' AND status='active'" ).fetchone()['n'] assert active_ns == 2, active_ns # the paired twin's URL is preserved on the canonical's desc desc = conn.execute('SELECT desc FROM listings WHERE id=?', (nb['dup_of'],)).fetchone()['desc'] or '' conn.close() assert 'also listed' in desc finally: shutil.rmtree(d, ignore_errors=True) def test_dedupe_reversible(): # A dup can be flipped active by hand; the next import re-derives it. d, db_path = fresh_db() try: state_dir = os.path.join(d, 'state') twin = make_item('2021 Indian Scout Bobber', price_num=6500, km='8000') write_states(state_dir, make_state({'ns1': twin}), make_state({'nb1': dict(twin)})) imp.run_import(state_dir, db_path) conn = db.connect(db_path) db.init_db(conn) nb_id = conn.execute( "SELECT id FROM listings WHERE region='nb'").fetchone()['id'] conn.execute("UPDATE listings SET status='active', dup_of=NULL WHERE id=?", (nb_id,)) conn.commit() assert conn.execute( "SELECT COUNT(*) AS n FROM listings WHERE status='dup'" ).fetchone()['n'] == 0 assert conn.execute( "SELECT COUNT(*) AS n FROM listings WHERE status='active'" ).fetchone()['n'] == 2 conn.close() imp.run_import(state_dir, db_path) # re-import re-pairs it conn = db.connect(db_path) db.init_db(conn) dup, active = (conn.execute( "SELECT COUNT(*) AS n FROM listings WHERE status='dup'").fetchone()['n'], conn.execute( "SELECT COUNT(*) AS n FROM listings WHERE status='active'").fetchone()['n']) conn.close() assert (dup, active) == (1, 1), (dup, active) finally: shutil.rmtree(d, ignore_errors=True) TESTS = [ ('gate rejections', test_gate_rejections), ('gate abort leaves DB untouched', test_gate_abort_leaves_db), ('dedupe cross-province twins', test_dedupe_twins), ('dedupe false-positive (same province)', test_dedupe_false_positive_same_province), ('dedupe false-positive (price differs)', test_dedupe_false_positive_price_diff), ('dedupe multi-unit pairing (min(ns,nb))', test_dedupe_multiunit_pairing), ('dedupe reversibility (flip + re-import)', test_dedupe_reversible), ('scoring spot-checks', test_scoring), ('gone after 2 misses (unit)', test_gone_after_two_misses), ('gone after 2 misses (full flow)', test_gone_full_flow), ] PAYLOAD_KEYS = {'id', 'title', 'price_display', 'price_num', 'km', 'location', 'region', 'url', 'img', 'desc', 'family', 'score', 'year', 'seller_type', 'extras', 'decision', 'first_seen', 'n_photos', 'last_photo_idx'} # B1 payload additions # P3 additive payload fields (UI-SPEC §10.7) P3_PAYLOAD_KEYS = {'points', 'points_breakdown', 'tank', 'slot', 'round_next'} REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) LIVE_DB = os.environ.get('BT_TEST_DB') or os.path.join( REPO_ROOT, 'data', 'bike-tinder.sqlite3') if not os.path.exists(LIVE_DB): # B1 test-arena copy (the brief's "copy DB"): pristine snapshot of the # live DB, never written by selfcheck (tests copy it further). _b1_copy = '/var/home/admin/dev/bike-tinder-b1/test-copy.sqlite3' if os.path.exists(_b1_copy): LIVE_DB = _b1_copy B1_STATE_CANDIDATES = ('/var/home/admin/dev/bike-tinder-b1', os.path.expanduser( '~/.hermes/scripts/motorcycle-hunt.d')) # --- P2 API checks (need the venv: fastapi + httpx). Run against a TEMP COPY # --- of the live DB — the live file is never opened by these tests. --------- try: from fastapi.testclient import TestClient from .api import create_app API_DEPS = True except ImportError: API_DEPS = False def api_env(): """Temp dir + TestClient over a copy of the live DB.""" if not os.path.exists(LIVE_DB): raise RuntimeError('live DB not found (%s) — run the import first' % LIVE_DB) d = tempfile.mkdtemp(prefix='bt-api-') db_path = os.path.join(d, 'test.sqlite3') shutil.copy2(LIVE_DB, db_path) return d, TestClient(create_app(db_path, os.path.join(d, 'img'))) def test_api_root(): d, c = api_env() try: r = c.get('/') assert r.status_code == 200, r.status_code assert 'Bike-Tinder' in r.text assert 'no-cache' in r.headers.get('cache-control', '') for path in ('/manifest.json', '/sw.js'): assert c.get(path).status_code == 200, path for name in ('icon-192.png', 'icon-512.png', 'icon-512-maskable.png', 'apple-touch-icon.png'): assert c.get('/icons/' + name).status_code == 200, name finally: shutil.rmtree(d, ignore_errors=True) def test_api_stats_health(): d, c = api_env() try: import sqlite3 conn = sqlite3.connect(os.path.join(d, 'test.sqlite3')) active = conn.execute( "SELECT COUNT(*) AS n FROM listings WHERE status='active'").fetchone()[0] total = conn.execute('SELECT COUNT(*) AS n FROM listings').fetchone()[0] expect_new = conn.execute( """SELECT COUNT(*) AS n FROM listings WHERE status='active' AND COALESCE(first_seen,'') > COALESCE( (SELECT value FROM prefs WHERE key='badge_seen_at'), '')""" ).fetchone()[0] conn.close() s = c.get('/api/stats').json() assert s['live'] == active, (s['live'], active) assert set(s) == {'live', 'fresh', 'new', 'likes', 'maybes', 'nopes', 'last_import'} # v1.1 item 7: + fresh # fresh = actives with no swipe row (the deck's new-pool predicate) conn = sqlite3.connect(os.path.join(d, 'test.sqlite3')) expect_fresh = conn.execute( "SELECT COUNT(*) AS n FROM listings l" " WHERE l.status='active'" " AND NOT EXISTS (SELECT 1 FROM swipes s" " WHERE s.listing_id = l.id)").fetchone()[0] conn.close() assert s['fresh'] == expect_fresh, (s['fresh'], expect_fresh) assert s['last_import'] is not None and s['last_import']['ok'] is True assert s['new'] == expect_new, (s['new'], expect_new) h = c.get('/api/health').json() assert h['ok'] is True assert h['counts']['active'] == active and h['counts']['total'] == total finally: shutil.rmtree(d, ignore_errors=True) def test_api_deck(): d, c = api_env() try: # seed a nope on today's top row: the deck must not serve it again p1 = c.get('/api/deck?limit=40').json() top = p1['items'][0] assert top['decision'] is None assert c.post('/api/swipe', json={'listing_id': top['id'], 'action': 'nope'}).status_code == 200 p1 = c.get('/api/deck?limit=40').json() items = p1['items'] assert items and len(items) == 40 assert all(it['id'] != top['id'] for it in items) # first row == highest (score, first_seen, id) among undecided active import sqlite3 conn = sqlite3.connect(os.path.join(d, 'test.sqlite3')) conn.row_factory = sqlite3.Row expect = conn.execute( """SELECT l.* FROM listings l WHERE l.status='active' AND NOT EXISTS (SELECT 1 FROM swipes s WHERE s.listing_id=l.id) ORDER BY l.score DESC, COALESCE(l.first_seen,'') DESC, l.id DESC LIMIT 1""").fetchone() conn.close() assert items[0]['id'] == expect['id'], (items[0]['id'], expect['id']) # monotonic page 1 + only undecided + payload contract prev = None for it in items: assert set(it.keys()) == PAYLOAD_KEYS | P3_PAYLOAD_KEYS, \ set(it.keys()) ^ (PAYLOAD_KEYS | P3_PAYLOAD_KEYS) assert it['decision'] is None if 'tank' in PAYLOAD_KEYS | P3_PAYLOAD_KEYS: assert isinstance(it['tank'], dict) and \ it['tank'].get('band') in ('like', 'maybe', 'nope') tup = (it['score'], it['first_seen'] or '', it['id']) if prev: assert tup <= prev, (tup, prev) prev = tup # cursor walk: zero overlap, monotonic across boundaries, terminates seen = {i['id'] for i in items} cursor = p1['next_cursor'] pages = 1 while cursor: page = c.get('/api/deck?limit=40&cursor=' + cursor).json() got = [i['id'] for i in page['items']] assert not (set(got) & seen), 'page overlap' for it in page['items']: assert it['decision'] is None tup = (it['score'], it['first_seen'] or '', it['id']) assert tup <= prev, 'cross-boundary ordering' prev = tup seen |= set(got) cursor = page['next_cursor'] pages += 1 assert pages < 100, 'cursor walk did not terminate' # every remaining undecided row must have appeared exactly once import sqlite3 as _sq conn2 = _sq.connect(os.path.join(d, 'test.sqlite3')) want = conn2.execute( """SELECT l.id FROM listings l WHERE l.status='active' AND NOT EXISTS (SELECT 1 FROM swipes s WHERE s.listing_id=l.id)""" ).fetchall() conn2.close() assert seen == ({r[0] for r in want} - {top['id']}), len(seen) finally: shutil.rmtree(d, ignore_errors=True) def test_api_swipe_lifecycle(): d, c = api_env() try: it = c.get('/api/deck?limit=5').json()['items'][0] lid = it['id'] r = c.post('/api/swipe', json={'listing_id': lid, 'action': 'like'}) assert r.status_code == 200 and r.json()['action'] == 'like' ids = [x['id'] for x in c.get('/api/list?decision=like&limit=500').json()['items']] assert lid in ids, 'swiped row must appear in decision=like' # re-POST edits (upsert, not append) r = c.post('/api/swipe', json={'listing_id': lid, 'action': 'nope'}) assert r.status_code == 200 and r.json()['action'] == 'nope' assert lid not in [x['id'] for x in c.get( '/api/list?decision=like&limit=500').json()['items']] assert lid in [x['id'] for x in c.get( '/api/list?decision=nope&limit=500').json()['items']] # DELETE clears; idempotent r = c.delete('/api/swipe/%d' % lid) assert r.status_code == 200 and r.json()['cleared'] is True assert lid not in [x['id'] for x in c.get( '/api/list?decision=nope&limit=500').json()['items']] assert lid in [x['id'] for x in c.get( '/api/list?decision=undecided&limit=500').json()['items']] r = c.delete('/api/swipe/%d' % lid) assert r.json()['cleared'] is False # invalid payloads rejected loudly assert c.post('/api/swipe', json={'listing_id': lid, 'action': 'wat'}).status_code == 400 assert c.post('/api/swipe', json={'listing_id': 999999999, 'action': 'like'}).status_code == 404 finally: shutil.rmtree(d, ignore_errors=True) def test_api_img_routes(): d, c = api_env() try: import sqlite3 conn = sqlite3.connect(os.path.join(d, 'test.sqlite3')) conn.row_factory = sqlite3.Row has_img = [r['id'] for r in conn.execute( 'SELECT id FROM listings WHERE img_url IS NOT NULL ORDER BY id LIMIT 2')] no_img = conn.execute( 'SELECT id FROM listings WHERE img_url IS NULL LIMIT 1').fetchone()['id'] conn.close() warm, cold302 = has_img[0], has_img[1] if len(has_img) > 1 else None img_dir = os.path.join(d, 'img') os.makedirs(img_dir, exist_ok=True) with open(os.path.join(img_dir, '%d.webp' % warm), 'wb') as fh: fh.write(b'RIFF\x00\x00\x00\x00WEBPpadding') r = c.get('/img/%d' % warm) assert r.status_code == 200, r.status_code assert r.headers.get('content-type', '').startswith('image/webp') r = c.get('/img/%d' % no_img) assert r.status_code == 404 and r.json()['detail'] == 'no image' if cold302 is not None: r = c.get('/img/%d' % cold302) if r.status_code == 302: assert r.headers.get('location', '').startswith('http') elif r.status_code == 404 and r.json().get('detail') == 'Not Found': # Known framework artifact: this fastapi/httpx pair's # TestClient mangles RedirectResponse into a bare 404. Real # uvicorn returns the true 302 (verified in the proof run). pass else: raise AssertionError('unexpected /img status %s' % r.status_code) assert c.get('/img/abc').status_code in (404, 422) finally: shutil.rmtree(d, ignore_errors=True) def test_api_badge_roundtrip(): d, c = api_env() try: before = c.get('/api/stats').json()['new'] r = c.post('/api/badge/seen') assert r.status_code == 200 and r.json()['badge_seen_at'] assert c.get('/api/stats').json()['new'] == 0, (before,) finally: shutil.rmtree(d, ignore_errors=True) # --- P3 checks (temp copies of the live DB; the live file is never opened) -- def p3_env(): """Temp copy of the live DB + TestClient + the copy's path.""" if not os.path.exists(LIVE_DB): raise RuntimeError('live DB not found (%s)' % LIVE_DB) d = tempfile.mkdtemp(prefix='bt-p3-') db_path = os.path.join(d, 'test.sqlite3') shutil.copy2(LIVE_DB, db_path) return d, TestClient(create_app(db_path, os.path.join(d, 'img'))), db_path def _sq_conn(db_path): import sqlite3 conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row return conn def test_p3_event_seeding_idempotent(): d, c, db_path = p3_env() try: conn = _sq_conn(db_path) n_swipes = conn.execute('SELECT COUNT(*) AS n FROM swipes').fetchone()['n'] tables = {r[0] for r in conn.execute( "SELECT name FROM sqlite_master WHERE type='table'")} n_events_before = 0 if 'swipe_events' in tables: n_events_before = conn.execute( 'SELECT COUNT(*) AS n FROM swipe_events').fetchone()['n'] conn.close() conn = db.connect(db_path) db.init_db(conn) n1 = conn.execute( 'SELECT COUNT(*) AS n FROM swipe_events').fetchone()['n'] if n_events_before == 0: # fresh pre-P3 DB: seeded exactly once from swipes rows assert n1 == n_swipes, (n1, n_swipes) seeded = conn.execute( 'SELECT listing_id, round, created_at FROM swipe_events LIMIT 1' ).fetchone() assert seeded['round'] == 1 match = conn.execute( 'SELECT created_at FROM swipes WHERE listing_id=?', (seeded['listing_id'],)).fetchone() assert seeded['created_at'] == match['created_at'] else: # migrated DB that already carries events (cleared swipes keep # their history) — seeding must not fire, events stay as found assert n1 == n_events_before, (n1, n_events_before) # second init does not reseed/duplicate db.init_db(conn) assert conn.execute( 'SELECT COUNT(*) AS n FROM swipe_events').fetchone()['n'] == n1 # a new swipe row after seeding is NOT re-seeded (only-when-empty) conn.execute( """INSERT INTO swipes (listing_id, action) SELECT id, 'like' FROM listings WHERE id NOT IN (SELECT listing_id FROM swipes) AND status='active' LIMIT 1""") conn.commit() db.init_db(conn) assert conn.execute( 'SELECT COUNT(*) AS n FROM swipe_events').fetchone()['n'] == n1 conn.close() finally: shutil.rmtree(d, ignore_errors=True) def test_p3_round_math_and_clear(): d, c, db_path = p3_env() try: s = c.get('/api/section').json() # first-read cold compose assert s['section']['status'] == 'open' it = [i for i in s['items'] if i['decision'] is None][0] lid, sec = it['id'], s['section']['id'] r1 = c.post('/api/swipe', json={'listing_id': lid, 'action': 'like', 'section_id': sec}).json() assert r1['round'] == 1 and r1['section_id'] == sec r2 = c.post('/api/swipe', json={'listing_id': lid, 'action': 'nope', 'section_id': sec}).json() assert r2['round'] == 2 r3 = c.post('/api/swipe', json={'listing_id': lid, 'action': 'nope'}).json() # list edit assert r3['round'] == 3 and r3['section_id'] is None conn = _sq_conn(db_path) evs = conn.execute( 'SELECT action, round, section_id FROM swipe_events' ' WHERE listing_id=? ORDER BY id', (lid,)).fetchall() shapes = [(e['action'], e['round'], e['section_id']) for e in evs] assert shapes[0] == ('like', 1, sec), shapes assert shapes[1] == ('nope', 2, sec), shapes assert shapes[2] == ('nope', 3, None), shapes # clear: UI verdict gone, event history kept r = c.delete('/api/swipe/%d' % lid) assert r.status_code == 200 and r.json()['cleared'] is True items = c.get('/api/list?decision=all&limit=500').json()['items'] row = [i for i in items if i['id'] == lid][0] assert row['decision'] is None assert conn.execute( 'SELECT COUNT(*) AS n FROM swipe_events WHERE listing_id=?', (lid,)).fetchone()['n'] == 3 # re-decide after clear -> round 4 (stored true round) r4 = c.post('/api/swipe', json={'listing_id': lid, 'action': 'maybe'}).json() assert r4['round'] == 4 conn.close() # engine rule: round>=2 weighs exactly 2x round 1 (same recency) from . import taste now = datetime.datetime(2026, 11, 11, 0, 0, 0) w1 = taste.event_weight('like', 1, '2026-11-10T00:00:00Z', now) w2 = taste.event_weight('like', 2, '2026-11-10T00:00:00Z', now) assert abs(w2 - 2.0 * w1) < 1e-9, (w1, w2) w3 = taste.event_weight('like', 9, '2026-11-10T00:00:00Z', now) assert abs(w3 - w2) < 1e-9 # rounds cap at 2 finally: shutil.rmtree(d, ignore_errors=True) def test_p3_section_invariants(): d, c, db_path = p3_env() try: from . import taste s = c.get('/api/section').json() assert {'section', 'items', 'progress', 'exhausted', 'mode', 'can_forge', 'counts', 'read_info'} <= set(s), set(s) items = s['items'] assert len(items) == 25 and s['section']['size'] == 25 by_slot = {} for it in items: by_slot.setdefault(it['slot'], []).append(it) assert it['slot'] in ('new', 'return', 'wildcard') assert len(by_slot['wildcard']) == 2 assert len(by_slot['return']) <= 5 assert sum(1 for i in by_slot['return'] if i['decision'] == 'nope') <= 2 assert len(by_slot['return']) + len(by_slot['wildcard']) + \ len(by_slot['new']) == 25 # returns carry a verdict and come after every new item, maybe>like>nope roles = ['new', 'return', 'wildcard'] seen = [] for it in items: seen.append(it['slot']) if it['slot'] == 'return': assert it['decision'] is not None first_return = seen.index('return') if 'return' in seen else None if first_return is not None: assert all(x == 'new' for x in seen[:first_return]) assert all(x != 'wildcard' for x in seen[:seen.index('wildcard', first_return)]) ret_roles = [i['decision'] for i in by_slot['return']] order = {'maybe': 0, 'like': 1, 'nope': 2} assert ret_roles == sorted(ret_roles, key=lambda a: order[a]), \ ret_roles # family cap <=4 (only non-NULL families are capped) fam = {} for it in items: if it['family']: fam[it['family']] = fam.get(it['family'], 0) + 1 assert all(v <= 4 for v in fam.values()), fam # unique positions 1..25, pinned order == payload order conn = _sq_conn(db_path) pos = [r['position'] for r in conn.execute( 'SELECT position FROM section_items WHERE section_id=?', (s['section']['id'],))] assert sorted(pos) == list(range(1, 26)), pos[:10] # >=3 price bands (the pool allows — 563 active rows) bands = {taste.price_band(it['price_num']) for it in items} assert len(bands - {None}) >= 3, bands # wildcards are unseen + not in any prior section (fresh copy) for w in by_slot['wildcard']: n = conn.execute( 'SELECT COUNT(*) AS n FROM swipe_events WHERE listing_id=?', (w['id'],)).fetchone()['n'] assert n == 0, (w['id'], n) # undecided items at read show decision None assert all(it['decision'] is None for it in items if it['slot'] == 'new') # size override forge composes exactly that size f = c.post('/api/section/forge', json={'size': 12}).json() assert len(f['items']) == 12 and f['section']['size'] == 12 slots = {} for it in f['items']: slots.setdefault(it['slot'], []).append(it) assert len(slots['wildcard']) == min(2, 12) assert len(slots['return']) <= 5 assert sum(len(v) for v in slots.values()) == 12 conn.close() finally: shutil.rmtree(d, ignore_errors=True) def test_p3_exhaust_and_report(): d, c, db_path = p3_env() try: c.post('/api/prefs', json={'key': 'forge_mode', 'value': 'ondemand'}) f = c.post('/api/section/forge', json={'size': 6}).json() sid = f['section']['id'] # v1.1 (item 5): EVERY item needs an in-section strike to complete # the section — returns re-enter the deck and are re-struck too. for it in f['items']: r = c.post('/api/swipe', json={'listing_id': it['id'], 'action': 'like', 'section_id': sid}) assert r.status_code == 200 s = c.get('/api/section').json() assert s['exhausted'] is True assert s['section']['status'] == 'report' assert s['progress'] == {'decided': 6, 'undecided': 0, 'total': 6}, \ s['progress'] assert s['counts']['likes'] + s['counts']['maybes'] + \ s['counts']['nopes'] == 6, s['counts'] assert s['can_forge'] is True and s['mode'] == 'ondemand' finally: shutil.rmtree(d, ignore_errors=True) def test_p3_forge_modes_strict_409(): d, c, db_path = p3_env() try: s = c.get('/api/section').json() # composed on profile None r = c.post('/api/prefs', json={'key': 'forge_mode', 'value': 'strict'}) assert r.status_code == 200 and r.json()['value'] == 'strict' fr = c.post('/api/section/forge', json={'size': 10}) assert fr.status_code == 409, fr.status_code ar = c.post('/api/agent/run', json={'source': 'mock'}) assert ar.status_code == 200 and ar.json()['source'] == 'mock' v = ar.json()['profile_version'] fr = c.post('/api/section/forge', json={'size': 10}) assert fr.status_code == 200, fr.text f = fr.json() assert f['section']['profile_version'] == v assert 'strict' in (f.get('note') or '') # ondemand forge runs a (mock) pass itself: version moves again c.post('/api/prefs', json={'key': 'forge_mode', 'value': 'ondemand'}) f2 = c.post('/api/section/forge', json={'size': 10}).json() assert f2['section']['profile_version'] > v, \ (f2['section']['profile_version'], v) assert 'pass mock' in (f2.get('note') or '') # live without a key refuses loudly assert c.post('/api/agent/run', json={'source': 'live'}).status_code == 400 finally: shutil.rmtree(d, ignore_errors=True) def test_p3_trends_shape_and_stats(): d, c, db_path = p3_env() try: t = c.get('/api/trends').json() assert set(t) == {'profile', 'stats', 'last_run', 'cold_start', 'forge_mode'}, set(t) st = t['stats'] assert {'decided', 'likes', 'maybes', 'nopes', 'hit_rate', 'hit_rate_plus_maybe', 'returns_placed', 'returns_re_struck', 'wildcards_placed', 'wildcards_hit', 'flips'} <= set(st), set(st) if st['decided']: assert st['likes'] + st['maybes'] + st['nopes'] == st['decided'] assert abs(st['hit_rate'] - st['likes'] / st['decided']) < 1e-9 else: assert st['hit_rate'] is None finally: shutil.rmtree(d, ignore_errors=True) def test_p3_agentpass_mock_and_fallback(): from . import agentpass as ag d, _c, db_path = p3_env() try: res = ag.run_pass(db_path, source='mock') assert res['ok'] is True and res['source'] == 'mock' assert res['profile_version'] >= 1 ag.validate_payload(res['payload']) conn = _sq_conn(db_path) run = conn.execute( 'SELECT ok, source, profile_version FROM agent_runs' ' ORDER BY id DESC LIMIT 1').fetchone() assert run['ok'] == 1 and run['source'] == 'mock' assert run['profile_version'] == res['profile_version'] conn.close() # deterministic: no randomness anywhere — same DB state + fixed clock # yields identical summary and payload on repeated calls now = datetime.datetime(2026, 11, 12, 0, 0, 0) s1 = ag.build_summary(db.connect(db_path), now=now) s2 = ag.build_summary(db.connect(db_path), now=now) assert s1 == s2, 'strike summary must be deterministic' p1 = ag.synthesize_payload(db.connect(db_path), now=now) p2 = ag.synthesize_payload(db.connect(db_path), now=now) assert p1 == p2, 'mock payload must be deterministic' ag.validate_payload(p1) # fallback: live attempt fails -> deterministic fallback profile. # (The copy may carry profiles from earlier live runs — scope the # assertion to the versions created by THIS test.) new_min = res['profile_version'] os.environ['OPENROUTER_API_KEY'] = 'test-key-never-used' orig = ag._live_call ag._live_call = lambda model, summary: (None, None, 'boom', None) try: res3 = ag.run_pass(db_path, source='live') finally: ag._live_call = orig del os.environ['OPENROUTER_API_KEY'] assert res3['source'] == 'fallback' and res3['ok'] is False assert res3['profile_version'] > res['profile_version'] ag.validate_payload(res3['payload']) conn = _sq_conn(db_path) sources = [r['source'] for r in conn.execute( 'SELECT source FROM taste_profile WHERE version >= ?' ' ORDER BY version', (new_min,))] assert sources == ['mock', 'fallback'], sources conn.close() finally: shutil.rmtree(d, ignore_errors=True) def test_p3_points_and_band(): from . import taste now = datetime.datetime(2026, 11, 10, 12, 0, 0) def L(i, title='2021 Indian Scout Bobber', desc='MVI done, red', family=None, score=0, price=None, km=None, year=2021, seller=None, extras=None, region='ns'): return {'id': i, 'title': title, 'desc': desc, 'family': family, 'score': score, 'price_num': price, 'km': km, 'year': year, 'seller_type': seller, 'extras': extras, 'region': region} fam = 'Indian Scout — top priority (sport×cruise)' listings = [ L(1, family=fam, score=91), L(2, '2018 Vulcan 900 Classic', 'black', family='Vulcan 900 Classic — the dream spec', score=73, price=5200, km=20000, year=2018), L(3, '2015 C50', family='Boulevard C50', score=50, price=4300, km=88000, year=2015), L(4, '2019 Rebel 500', family='Rebel 500', score=50, price=6100, km=4000, year=2019), ] events = [ {'listing_id': 1, 'action': 'like', 'round': 1, 'created_at': '2026-11-05T12:00:00Z'}, {'listing_id': 2, 'action': 'like', 'round': 1, 'created_at': '2026-11-05T12:00:00Z'}, {'listing_id': 1, 'action': 'like', 'round': 2, 'created_at': '2026-11-08T12:00:00Z'}, {'listing_id': 3, 'action': 'nope', 'round': 1, 'created_at': '2026-11-05T12:00:00Z'}, {'listing_id': 4, 'action': 'maybe', 'round': 1, 'created_at': '2026-11-05T12:00:00Z'}, ] eng = taste.TasteEngine(listings, events, None, now=now) assert eng.w == min(0.85, 5 / 60.0) assert eng.points(1) == 91 # known answers for these exact fixtures assert eng.points(2) == 70 assert eng.points(3) == 47 assert eng.points(4) == 51 # zero events: w=0 -> points == rubric score z = taste.TasteEngine(listings, [], None, now=now) assert z.w == 0.0 assert all(z.points(l['id']) == l['score'] for l in listings) # zero profile: engine identical to null adjustments assert z.agent_adj(1) == 0.0 # band gates assert taste.TasteEngine.band(29) == 'nope' assert taste.TasteEngine.band(30) == 'maybe' assert taste.TasteEngine.band(70) == 'maybe' assert taste.TasteEngine.band(71) == 'like' assert taste.TasteEngine.band(None) is None # ramp caps at 0.85 with enough events many = [{'listing_id': 1, 'action': 'like', 'round': 1, 'created_at': '2026-11-10T12:00:00Z'}] * 120 big = taste.TasteEngine(listings, many, None, now=now) assert big.w == 0.85 # adjustments clamp at +/-10 per listing prof = {'adjustments': [ {'feature': 'family', 'value': fam, 'delta': 7}, {'feature': 'color', 'value': 'red', 'delta': 5}, ]} c_eng = taste.TasteEngine(listings, events, prof, now=now) assert c_eng.agent_adj(1) == 10.0 # breakdown is source-labeled br = eng.breakdown(1) assert {b['source'] for b in br} <= {'rubric', 'taste', 'agent'} assert br[0]['source'] == 'rubric' # regression (#3): a listing OUTSIDE the engine's pool (delisted # mid-batch, never swiped) must degrade quietly, not crash the render assert eng.points(9999) is None assert eng.breakdown(9999) == [] def test_p3_payload_additions(): d, c, db_path = p3_env() try: lst = c.get('/api/list?limit=5').json()['items'] assert lst for it in lst: assert set(it.keys()) == PAYLOAD_KEYS | P3_PAYLOAD_KEYS assert it['tank'] and it['tank']['band'] in ('like', 'maybe', 'nope') assert it['round_next'] in (1, 2) assert it['slot'] is None one = c.get('/api/listing/%d' % lst[0]['id']).json() assert isinstance(one['points_breakdown'], list) and \ one['points_breakdown'] assert all('source' in l for l in one['points_breakdown']) sec = c.get('/api/section').json() for it in sec['items']: assert it['slot'] in ('new', 'return', 'wildcard') assert isinstance(it['points_breakdown'], list) finally: shutil.rmtree(d, ignore_errors=True) def test_issue3_section_survives_midsection_delist(): """#3 regression: one card delisted after compose must not 500 the deck. The section render asks the taste engine for a breakdown of every card. A listing that left the active pool mid-batch AND has no swipe history is not in the engine's by_id map; the unguarded lookup used to raise, killing GET /api/section and GET /api/listing/{id} with a 500 (live incident 2026-11-15: section 57 + delisted listing 298). """ d, c, db_path = p3_env() try: sec = c.get('/api/section') assert sec.status_code == 200, sec.status_code body = sec.json() conn = _sq_conn(db_path) ev_ids = {r['listing_id'] for r in conn.execute( 'SELECT DISTINCT listing_id FROM swipe_events')} it = next((x for x in body['items'] if x['id'] not in ev_ids), None) assert it is not None, 'fixture section has no unswiped card' conn.execute("UPDATE listings SET status='gone' WHERE id=?", (it['id'],)) conn.commit() conn.close() r = c.get('/api/section') assert r.status_code == 200, r.status_code assert it['id'] in [x['id'] for x in r.json()['items']] one = c.get('/api/listing/%d' % it['id']) assert one.status_code == 200, one.status_code assert one.json()['points_breakdown'] == [] finally: shutil.rmtree(d, ignore_errors=True) # --- B1 checks (photo experience v2, UI-SPEC §11 / DATA-MODEL B1) ----------- def make_b1_item(title, imgs=None, img=None, price='$5,000', km='8000'): it = make_item(title, price=price, km=km) if img is not None: it['img'] = img if imgs is not None: it['imgs'] = imgs return it def test_b1_migration_guarded(): d, db_path = fresh_db() try: conn = db.connect(db_path) db.init_db(conn) lc = {r['name'] for r in conn.execute( "PRAGMA table_info('listings')")} sc = {r['name'] for r in conn.execute("PRAGMA table_info('swipes')")} assert 'imgs' in lc and 'last_photo_idx' in sc, (lc, sc) db.init_db(conn) # second pass: idempotent lc2 = {r['name'] for r in conn.execute( "PRAGMA table_info('listings')")} assert lc2 == lc conn.close() finally: shutil.rmtree(d, ignore_errors=True) def test_b1_importer_gallery_carry(): """imgs -> listings.imgs (JSON, page order); img_url keeps `img` and falls back to imgs[0]; empty/junk galleries -> NULL; re-import idempotent.""" d, db_path = fresh_db() try: state_dir = os.path.join(d, 'state') CDN = 'https://media.kijiji.ca/api/v1/x/%s?rule=kijijica-640-webp' n8 = [CDN % ('a%d' % i) for i in range(8)] n3 = [CDN % ('b%d' % i) for i in range(3)] ns = { 'r8': make_b1_item('2021 Indian Scout Bobber 8shot', imgs=n8), 'r3': make_b1_item('2021 Indian Scout Bobber 3shot', imgs=n3, img=n3[0]), 'empty': make_b1_item('Kawasaki Vulcan 900 Custom', imgs=[]), 'plain': make_b1_item('Honda Shadow'), 'junk': make_b1_item('Suzuki C50', imgs='not-a-list'), } nb = {'other': make_item('Some Other Bike', price_num=9999)} write_states(state_dir, make_state(ns), make_state(nb)) FIXED = '2026-11-11T12:00:00Z' orig = imp.now_iso imp.now_iso = lambda: FIXED try: r1 = imp.run_import(state_dir, db_path) r2 = imp.run_import(state_dir, db_path) finally: imp.now_iso = orig assert r1['ok'] and r2['ok'] and r2['total'] == 6 and r2['new'] == 0, \ (r1, r2) conn = db.connect(db_path) db.init_db(conn) row = lambda s: conn.execute( 'SELECT * FROM listings WHERE src_id=?', (s,)).fetchone() r8 = row('r8') assert json.loads(r8['imgs']) == n8 and r8['img_url'] == n8[0] r3 = row('r3') assert json.loads(r3['imgs']) == n3 and r3['img_url'] == n3[0] assert row('empty')['imgs'] is None and row('empty')['img_url'] is None plain = row('plain') assert plain['imgs'] is None and plain['img_url'] is None assert row('junk')['imgs'] is None before = [tuple(x) for x in conn.execute( 'SELECT * FROM listings ORDER BY id')] s_n = conn.execute( 'SELECT COUNT(*) AS n FROM swipes').fetchone()['n'] conn.close() imp.now_iso = lambda: FIXED try: r3 = imp.run_import(state_dir, db_path) finally: imp.now_iso = orig conn = db.connect(db_path) db.init_db(conn) after = [tuple(x) for x in conn.execute( 'SELECT * FROM listings ORDER BY id')] assert s_n == conn.execute( 'SELECT COUNT(*) AS n FROM swipes').fetchone()['n'] conn.close() assert before == after, 're-import must be byte-stable (fixed clock)' finally: shutil.rmtree(d, ignore_errors=True) def test_b1_640row_reimport_idempotence(): """Deliverable A: the 640-row copy DB re-imported from the hunt state snapshot stays row-for-row stable across two runs (fixed clock).""" sourced = None for cand in B1_STATE_CANDIDATES: if (os.path.exists(os.path.join(cand, 'kijiji-ns.json')) and os.path.exists(os.path.join(cand, 'kijiji-nb.json'))): sourced = cand break assert sourced, 'hunt state snapshot not found (%s)' % ( B1_STATE_CANDIDATES,) with open(os.path.join(sourced, 'kijiji-ns.json')) as fh: ns = json.load(fh) with open(os.path.join(sourced, 'kijiji-nb.json')) as fh: nb = json.load(fh) d = tempfile.mkdtemp(prefix='bt-b1-640-') try: state_dir = os.path.join(d, 'state') db_path = os.path.join(d, 'copy.sqlite3') shutil.copy2(LIVE_DB, db_path) ns['fetched_at'] = nb['fetched_at'] = recent_iso() # evergreen gates write_states(state_dir, ns, nb) FIXED = '2026-11-11T12:00:00Z' orig = imp.now_iso imp.now_iso = lambda: FIXED try: r1 = imp.run_import(state_dir, db_path) finally: imp.now_iso = orig assert r1['ok'], r1 conn = db.connect(db_path) db.init_db(conn) total = conn.execute('SELECT COUNT(*) AS n FROM listings').fetchone()['n'] active = conn.execute( "SELECT COUNT(*) AS n FROM listings WHERE status='active'" ).fetchone()['n'] dup = conn.execute( "SELECT COUNT(*) AS n FROM listings WHERE status='dup'" ).fetchone()['n'] before = [tuple(x) for x in conn.execute( 'SELECT * FROM listings ORDER BY id')] s_n = conn.execute('SELECT COUNT(*) AS n FROM swipes').fetchone()['n'] conn.close() assert total == 640 and active == 563 and dup == 77, \ (total, active, dup) # the B1 fixture is stable at 640 imp.now_iso = lambda: FIXED try: r2 = imp.run_import(state_dir, db_path) finally: imp.now_iso = orig assert r2['ok'] and r2['new'] == 0 and r2['total'] == 640, r2 conn = db.connect(db_path) db.init_db(conn) after = [tuple(x) for x in conn.execute( 'SELECT * FROM listings ORDER BY id')] s_n2 = conn.execute('SELECT COUNT(*) AS n FROM swipes').fetchone()['n'] conn.close() assert before == after, '640 rows must be byte-stable across import' assert s_n == s_n2, 'swipes must be untouched by re-import' finally: shutil.rmtree(d, ignore_errors=True) def test_b1_payload_fields_and_rotation_math(): """n_photos == len(imgs) (segments count = n) on deck/section/list rows; §11.5 rotation lead = (last_photo_idx + 1) % n for null / n=1 / n=3 / n=8.""" d, c, db_path = p3_env() try: _conn = db.connect(db_path) db.init_db(_conn) # B1 columns on the pristine copy _conn.close() conn = _sq_conn(db_path) CDN = 'https://media.kijiji.ca/api/v1/x/%s?rule=kijijica-640-webp' n8 = [CDN % ('g8-%d' % i) for i in range(8)] n3 = [CDN % ('g3-%d' % i) for i in range(3)] conn.execute( "INSERT INTO listings (region, src_id, title, url, status," " img_url, imgs, score, first_seen) VALUES" " ('ns','b1g8','B1 eight-photo','http://x/e','active',?,?,200," " '2026-11-11T00:00:00Z')", (n8[0], json.dumps(n8))) g8 = conn.execute("SELECT id FROM listings WHERE src_id='b1g8'" ).fetchone()['id'] conn.execute( "INSERT INTO listings (region, src_id, title, url, status," " img_url, imgs, score, first_seen) VALUES" " ('ns','b1g3','B1 three-photo','http://x/t','active',?,?,190," " '2026-11-11T00:00:01Z')", (n3[0], json.dumps(n3))) g3 = conn.execute("SELECT id FROM listings WHERE src_id='b1g3'" ).fetchone()['id'] # a swiped fixture row with a stored photo idx (rotation source) conn.execute( "INSERT INTO listings (region, src_id, title, url, status," " img_url, imgs, score, first_seen) VALUES" " ('ns','b1rot','B1 rotation','http://x/r','active',?,?,180," " '2026-11-11T00:00:02Z')", (n8[0], json.dumps(n8))) rot = conn.execute("SELECT id FROM listings WHERE src_id='b1rot'" ).fetchone()['id'] conn.execute( "INSERT INTO swipes (listing_id, action, last_photo_idx)" " VALUES (?, 'like', 2)", (rot,)) conn.commit() conn.close() # payload contract: n_photos mirrors the gallery length for payload in (c.get('/api/list?limit=500').json()['items'], c.get('/api/deck?limit=100').json()['items']): m = {i['id']: i for i in payload} for gid, want in ((g8, 8), (g3, 3), (rot, 8)): if gid in m: assert m[gid]['n_photos'] == want, (gid, m[gid]) one = c.get('/api/listing/%d' % rot).json() assert one['n_photos'] == 8 and one['last_photo_idx'] == 2 # rotation math on the real stored values (FE: (last+1) % n) def lead(li, n): if not n or n < 2 or li is None or li < 0: return 0 return (li + 1) % n assert lead(None, 8) == 0 # no stored index -> photo 0 assert lead(0, 1) == 0 # single photo -> always 0 assert lead(0, 3) == 1 and lead(2, 3) == 0 # 3-photo wrap assert lead(1, 8) == 2 and lead(7, 8) == 0 # 8-photo wrap finally: shutil.rmtree(d, ignore_errors=True) def test_b1_img_route_chain(): """GET /img/{id}?i=n: mirrored file (incl. {id}_0.webp as the gallery primary name) -> CDN 302 -> 404; i=0 keeps the legacy chain; out-of-range and unknown rows 404.""" d, c, db_path = p3_env() try: import sqlite3 as _sq _conn = db.connect(db_path) db.init_db(_conn) # B1 columns on the pristine copy _conn.close() conn = _sq.connect(db_path) conn.row_factory = _sq.Row CDN = 'https://media.kijiji.ca/api/v1/x/%s?rule=kijijica-640-webp' n3 = [CDN % ('c%d' % i) for i in range(3)] conn.execute( "INSERT INTO listings (region, src_id, title, url, status," " img_url, imgs) VALUES ('ns','b1img','B1 images','http://x/i'," " 'active',?,?)", (n3[0], json.dumps(n3))) gid = conn.execute("SELECT id FROM listings WHERE src_id='b1img'" ).fetchone()['id'] conn.execute( "INSERT INTO listings (region, src_id, title, url, status," " img_url) VALUES ('ns','b1leg','B1 legacy','http://x/l'," " 'active',?)", (CDN % 'L',)) lg = conn.execute("SELECT id FROM listings WHERE src_id='b1leg'" ).fetchone()['id'] conn.commit() conn.close() img_dir = os.path.join(d, 'img') for n in range(3): p = os.path.join(img_dir, '%d_%d.webp' % (gid, n)) with open(p, 'wb') as fh: fh.write(b'RIFF\x00\x00\x00\x00WEBP g%d n%d' % (gid, n)) # legacy chain: {id}.webp file wins for i=0 with open(os.path.join(img_dir, '%d.webp' % lg), 'wb') as fh: fh.write(b'RIFF\x00\x00\x00\x00WEBP legacy') r = c.get('/img/%d' % lg) assert r.status_code == 200 and r.headers.get( 'content-type', '').startswith('image/webp') # i=0 with no {id}.webp -> accepts the gallery primary {id}_0.webp r = c.get('/img/%d' % gid) assert r.status_code == 200, r.status_code # gallery extras answered for every n for n in (1, 2): r = c.get('/img/%d?i=%d' % (gid, n)) assert r.status_code == 200, (n, r.status_code) assert r.headers.get('content-type', '').startswith('image/webp') # out-of-range -> 404; negative -> validation error; unknown -> 404 assert c.get('/img/%d?i=3' % gid).status_code == 404 assert c.get('/img/%d?i=-1' % gid).status_code in (404, 422) assert c.get('/img/%d?i=1' % lg).status_code == 404 # legacy 1-photo assert c.get('/img/999999').status_code == 404 finally: shutil.rmtree(d, ignore_errors=True) def test_b1_swipe_photo_idx(): """POST /api/swipe optionally persists photo_idx; an absent key preserves the stored index; delete clears it (payload last_photo_idx -> null).""" d, c, db_path = p3_env() try: import sqlite3 as _sq _conn = db.connect(db_path) db.init_db(_conn) # B1 columns on the pristine copy _conn.close() conn = _sq.connect(db_path) conn.row_factory = _sq.Row CDN = 'https://media.kijiji.ca/api/v1/x/%s.jpg' n3 = [CDN % ('p%d' % i) for i in range(3)] conn.execute( "INSERT INTO listings (region, src_id, title, url, status," " img_url, imgs, score, first_seen) VALUES ('ns','b1sw'," " 'B1 swipe idx','http://x/s','active',?,?,9999," " '2026-11-11T00:00:00Z')", (n3[0], json.dumps(n3))) gid = conn.execute("SELECT id FROM listings WHERE src_id='b1sw'" ).fetchone()['id'] conn.commit() conn.close() r = c.post('/api/swipe', json={'listing_id': gid, 'action': 'maybe', 'photo_idx': 2}) assert r.status_code == 200 and r.json()['last_photo_idx'] == 2, r.text # absent photo_idx preserves the stored index r = c.post('/api/swipe', json={'listing_id': gid, 'action': 'nope'}) assert r.status_code == 200 and r.json()['last_photo_idx'] == 2, r.text # explicit overwrite (0 is a valid photo index) r = c.post('/api/swipe', json={'listing_id': gid, 'action': 'like', 'photo_idx': 0}) assert r.status_code == 200 and r.json()['last_photo_idx'] == 0, r.text # invalid values rejected loudly assert c.post('/api/swipe', json={'listing_id': gid, 'action': 'nope', 'photo_idx': -1}).status_code == 400 assert c.post('/api/swipe', json={'listing_id': gid, 'action': 'nope', 'photo_idx': 'x'}).status_code == 400 assert c.post('/api/swipe', json={'listing_id': gid, 'action': 'nope', 'photo_idx': True}).status_code == 400 # explicit null clears the index r = c.post('/api/swipe', json={'listing_id': gid, 'action': 'maybe', 'photo_idx': None}) assert r.status_code == 200 and r.json()['last_photo_idx'] is None # payload reflects the persisted index; delete removes the row c.post('/api/swipe', json={'listing_id': gid, 'action': 'like', 'photo_idx': 1}) lst = c.get('/api/list?limit=500').json()['items'] row = [i for i in lst if i['id'] == gid][0] assert row['n_photos'] == 3 and row['last_photo_idx'] == 1 c.delete('/api/swipe/%d' % gid) lst = c.get('/api/list?limit=500').json()['items'] row = [i for i in lst if i['id'] == gid][0] assert row['last_photo_idx'] is None and row['n_photos'] == 3 finally: shutil.rmtree(d, ignore_errors=True) # --- v1.1 checks (temp copies of the source DB only — see p3_env) ---------- def test_v11_item5_section_counter(): """Returns-only section must read 0/N decided until real in-section events exist; report counts and exhaustion key off those events.""" d, c, db_path = p3_env() try: conn = _sq_conn(db_path) conn.execute('DELETE FROM section_items') conn.execute("UPDATE sections SET status='closed'" " WHERE status!='closed'") conn.execute( """INSERT INTO swipes (listing_id, action, created_at) SELECT id, 'maybe', '2026-11-01T00:00:00Z' FROM listings WHERE status='active' AND id NOT IN (SELECT listing_id FROM swipes)""") conn.execute( """INSERT INTO swipe_events (listing_id, action, round, section_id) SELECT id, 'maybe', 1, NULL FROM listings WHERE status='active' AND id NOT IN (SELECT listing_id FROM swipe_events)""") conn.commit() conn.close() s = c.get('/api/section').json() # cold compose total = s['progress']['total'] assert total > 0, s assert s['recipe']['new'] == 0, s['recipe'] # every item wears a LIFETIME verdict but zero in-section decisions for it in s['items']: assert it['decision'] is not None, it['id'] assert it['decided_this_section'] is False, it['id'] assert s['progress']['decided'] == 0, s['progress'] sec_id = s['section']['id'] # list edit (no section_id) must not count as an in-section decision lid0 = s['items'][0]['id'] c.post('/api/swipe', json={'listing_id': lid0, 'action': 'like'}) assert c.get('/api/section').json()['progress']['decided'] == 0 # first real in-section swipe -> 1/total c.post('/api/swipe', json={'listing_id': lid0, 'action': 'like', 'section_id': sec_id}) s2 = c.get('/api/section').json() assert s2['progress']['decided'] == 1, s2['progress'] # second -> 2/total; report counts reflect in-section verdicts only lid1 = s2['items'][1]['id'] c.post('/api/swipe', json={'listing_id': lid1, 'action': 'maybe', 'section_id': sec_id}) s3 = c.get('/api/section').json() assert s3['progress']['decided'] == 2, s3['progress'] assert s3['counts']['likes'] == 1, s3['counts'] assert s3['counts']['maybes'] == 1, s3['counts'] # undo residual (Q5, accepted): DELETE clears the swipe row but the # section-stamped event stays -> count holds (clamp-only behavior) c.delete('/api/swipe/%d' % lid1) s4 = c.get('/api/section').json() assert s4['progress']['decided'] == 2, s4['progress'] c.post('/api/swipe', json={'listing_id': lid1, 'action': 'maybe', 'section_id': sec_id}) # strike the rest in-section -> report flips, counts match verdicts for it in s4['items']: if it['id'] not in (lid0, lid1): c.post('/api/swipe', json={'listing_id': it['id'], 'action': 'maybe', 'section_id': sec_id}) s5 = c.get('/api/section').json() assert s5['exhausted'] is True assert s5['section']['status'] == 'report' assert s5['progress']['decided'] == total, s5['progress'] assert s5['counts']['likes'] == 1, s5['counts'] assert s5['counts']['maybes'] == total - 1, s5['counts'] # a fresh returns-only section of the same shape composes cleanly # (lifetime verdicts present, in-section decided == 0 again) f = c.post('/api/section/forge', json={}).json() assert f['recipe']['new'] == 0, f['recipe'] assert f['progress']['decided'] == 0, f['progress'] for it in f['items']: assert it['decision'] is not None and not it['decided_this_section'] finally: shutil.rmtree(d, ignore_errors=True) def test_v11_item7_stats_fresh(): """Item 7: /api/stats.fresh counts actives with no swipe row and moves on swipe (the deck's new-pool predicate).""" d, c, db_path = p3_env() try: conn = _sq_conn(db_path) rows = conn.execute( "SELECT id FROM listings WHERE status='active' LIMIT 5" ).fetchall() assert len(rows) == 5 ids = [r['id'] for r in rows] conn.execute("UPDATE listings SET status='gone' WHERE status='active'") conn.execute("DELETE FROM swipes WHERE listing_id IN (%s)" % ','.join('?' * len(ids)), ids) conn.execute("DELETE FROM swipe_events WHERE listing_id IN (%s)" % ','.join('?' * len(ids)), ids) for lid in ids: conn.execute("UPDATE listings SET status='active' WHERE id=?", (lid,)) for lid in ids[:3]: conn.execute( "INSERT INTO swipes (listing_id, action, created_at)" " VALUES (?, 'maybe', '2026-11-01T00:00:00Z')", (lid,)) conn.commit() conn.close() st = c.get('/api/stats').json() assert st['live'] == 5, st assert st['fresh'] == 2, st # 2 of 5 still never-touched # swiping one of the untouched pair moves fresh to 1 on next read c.post('/api/swipe', json={'listing_id': ids[3], 'action': 'like'}) assert c.get('/api/stats').json()['fresh'] == 1 finally: shutil.rmtree(d, ignore_errors=True) def test_v11_item8_pool_and_dry(): """Item 8/6: pool counts ride section_state; 0/0/0 composes an empty section; a fresh=0, returns>0 catalog labels the forge note dry.""" d, c, db_path = p3_env() try: conn = _sq_conn(db_path) conn.execute('DELETE FROM section_items') conn.execute("UPDATE sections SET status='closed'" " WHERE status!='closed'") # every active gets a maybe swipe + round-1 event (returns alive, # fresh + wildcards drained) conn.execute( """INSERT INTO swipes (listing_id, action, created_at) SELECT id, 'maybe', '2026-11-01T00:00:00Z' FROM listings WHERE status='active' AND id NOT IN (SELECT listing_id FROM swipes)""") conn.execute( """INSERT INTO swipe_events (listing_id, action, round, section_id) SELECT id, 'maybe', 1, NULL FROM listings WHERE status='active' AND id NOT IN (SELECT listing_id FROM swipe_events)""") conn.commit() conn.close() s = c.get('/api/section').json() # dry but alive assert s['pool']['fresh'] == 0 and s['pool']['returns'] > 0, s['pool'] assert s['recipe']['new'] == 0 and len(s['items']) > 0 f = c.post('/api/section/forge', json={}).json() assert f['recipe']['new'] == 0 and len(f['items']) > 0 assert 'catalog fully reviewed' in (f.get('note') or ''), f.get('note') # drain returns too (one-shot) + wildcards: pool 0/0/0 -> empty compose conn = _sq_conn(db_path) conn.execute("DELETE FROM section_items") conn.execute("UPDATE sections SET status='closed'" " WHERE status!='closed'") # burn return eligibility: every active already had its return slot # -> re-insert one-shot return rows for all of them conn.execute( """INSERT INTO section_items (section_id, listing_id, slot, position) SELECT (SELECT id FROM sections ORDER BY id DESC LIMIT 1), listing_id, 'return', 1 FROM swipes WHERE listing_id IN (SELECT id FROM listings WHERE status='active')""") conn.commit() conn.close() s2 = c.get('/api/section').json() assert s2['pool'] == {'fresh': 0, 'returns': 0, 'wildcards': 0}, s2['pool'] assert s2['items'] == [] assert s2['recipe']['new'] == 0 and s2['recipe']['return'] == 0 assert s2['recipe']['wildcard'] == 0, s2['recipe'] finally: shutil.rmtree(d, ignore_errors=True) # --- item 9: batch formula v2 compose matrix (A1-A8, staged pools) -------- def _v11_wipe(conn): """Reset a temp copy to a controlled catalog: no sections/swipes/events, everything parked 'gone' (rows kept so ids/families can be re-used).""" conn.execute('DELETE FROM section_items') conn.execute('DELETE FROM sections') conn.execute('DELETE FROM swipes') conn.execute('DELETE FROM swipe_events') conn.execute("UPDATE listings SET status='gone' WHERE status!='dup'") def _v11_ids(conn, n, start=0): return [r['id'] for r in conn.execute( 'SELECT id FROM listings ORDER BY id LIMIT ? OFFSET ?', (n, start))] def _v11_activate(conn, ids, family=None, score=None): for lid in ids: if score is None: conn.execute("UPDATE listings SET status='active', family=?" " WHERE id=?", (family, lid)) else: conn.execute("UPDATE listings SET status='active', family=?," " score=? WHERE id=?", (family, score, lid)) def _v11_stage_returns(conn, ids, action='maybe', rests=()): """swipes row + round-1 event per id; rests get a round-2 nope event (the REST rule bars them from returns forever).""" rests = set(rests) for lid in ids: conn.execute("INSERT INTO swipes (listing_id, action, created_at)" " VALUES (?, ?, '2026-11-01T00:00:00Z')", (lid, action)) conn.execute("INSERT INTO swipe_events (listing_id, action, round," " section_id) VALUES (?, ?, 1, NULL)", (lid, action)) if lid in rests: conn.execute("INSERT INTO swipe_events (listing_id, action," " round, section_id) VALUES (?, 'nope', 2, NULL)", (lid,)) def _v11_stage_fresh(conn, ids, seen=False): """No swipes row (=fresh). seen=True drops wildcard eligibility via a round-1 event (fresh pool and wildcard pool are otherwise the same).""" for lid in ids: if seen: conn.execute("INSERT INTO swipe_events (listing_id, action," " round, section_id) VALUES (?, 'maybe', 1, NULL)", (lid,)) def _v11_stage_support(conn, ids, family): """Gone-status listings whose events tilt a family's net (nope-gate).""" for lid in ids: conn.execute("UPDATE listings SET family=? WHERE id=?", (family, lid)) conn.execute("INSERT INTO swipe_events (listing_id, action, round," " section_id) VALUES (?, 'like', 1, NULL)", (lid,)) def _v11_compose(db_path, size=25): from . import sections as secmod conn = db.connect(db_path) db.init_db(conn) engine = secmod.build_engine( conn, now=datetime.datetime.now(datetime.timezone.utc).replace( tzinfo=None)) sec = secmod.compose_section(conn, engine, size=size) recipe = json.loads(sec['recipe']) items = [dict(r) for r in secmod.section_items(conn, sec)] conn.commit() conn.close() return recipe, items, engine def _slots(items): out = {'new': [], 'return': [], 'wildcard': []} for it in items: out[it['slot']].append(it) return out def test_v11_item9_compose_matrix(): """A1-A8 of the item 9 acceptance criteria on deterministic pools.""" d, _c, db_path = p3_env() def env(): conn = _sq_conn(db_path) _v11_wipe(conn) conn.commit() return conn try: # --- A1: 30 fresh / 10 returns / wildcards -> 25: 21/2/2 ---------- conn = env() ids = _v11_ids(conn, 60) _v11_activate(conn, ids[0:28]) # 28 fresh _v11_stage_fresh(conn, ids[0:28]) # (all wild-eligible) _v11_activate(conn, ids[28:38]) # 10 returns _v11_stage_returns(conn, ids[28:38]) conn.commit(); conn.close() recipe, items, engine = _v11_compose(db_path) slots = _slots(items) assert len(items) == 25, len(items) assert (recipe['new'], recipe['return'], recipe['wildcard']) \ == (21, 2, 2), recipe assert (len(slots['new']), len(slots['return']), len(slots['wildcard'])) == (21, 2, 2) # pinned order: new -> return -> wildcard roles = [it['slot'] for it in items] assert roles == ['new'] * 21 + ['return'] * 2 + ['wildcard'] * 2, \ roles[:2] # fresh picks ordered points DESC (non-increasing) pts = [engine.points(it['id']) for it in slots['new']] assert all(a >= b for a, b in zip(pts, pts[1:])), pts # --- A2a: 5 fresh / 20 returns / wildcards -> 5/18/2 --------------- conn = env() ids = _v11_ids(conn, 60) _v11_activate(conn, ids[0:3]) _v11_stage_fresh(conn, ids[0:3]) # 3 fresh-only _v11_activate(conn, ids[3:5]) _v11_stage_fresh(conn, ids[3:5]) # 2 fresh+wild _v11_activate(conn, ids[5:25]) # 20 returns _v11_stage_returns(conn, ids[5:25]) conn.commit(); conn.close() recipe, items, _e = _v11_compose(db_path) assert len(items) == 25, len(items) # Wildcards are BY DEFINITION never-swiped (wild pool ⊆ fresh pool), # so on a 5-fresh day the 2 wilds come from those 5: the deck reads # 5 fresh-total (3 new + 2 wild) + 20 second looks = 25 — Glenn's # "5 fresh listed, rest second looks". The spec's "5 fresh + 18 # returns + 2 wild" arithmetic assumed 7 fresh-eligible listings; # with 5 the pool-honest fill is 20 (implementer-verified). assert (recipe['new'], recipe['return'], recipe['wildcard']) \ == (3, 20, 2), recipe # --- A2b: same but the wildcard pool is dry -> 5/20/0 -------------- conn = env() ids = _v11_ids(conn, 60) _v11_activate(conn, ids[0:5]) _v11_stage_fresh(conn, ids[0:5], seen=True) # fresh, no wilds _v11_activate(conn, ids[5:25]) # 20 returns _v11_stage_returns(conn, ids[5:25]) # conn.commit(); conn.close() recipe, items, _e = _v11_compose(db_path) assert len(items) == 25, len(items) assert (recipe['new'], recipe['return'], recipe['wildcard']) \ == (5, 20, 0), recipe # --- A3: 23 fresh / 5 returns / wildcards -> 21/2/2 (the sign-off # --- sentence shape) ---------------------------------------------- conn = env() ids = _v11_ids(conn, 60) _v11_activate(conn, ids[0:23]) _v11_stage_fresh(conn, ids[0:23]) _v11_activate(conn, ids[23:28]) # 5 returns _v11_stage_returns(conn, ids[23:28]) conn.commit(); conn.close() recipe, items, _e = _v11_compose(db_path) slots = _slots(items) assert len(items) == 25 and len(slots['new']) == 21 \ and len(slots['return']) == 2 \ and len(slots['wildcard']) == 2, recipe # --- A4: 0 fresh / 12 returns (2 nope) -> 12 returns, <=2 nopes, # --- LAST_TWO excl: second compose repeats nothing ----------------- conn = env() ids = _v11_ids(conn, 60) _v11_activate(conn, ids[0:12]) _v11_stage_returns(conn, ids[0:10], 'maybe') _v11_stage_returns(conn, ids[10:12], 'nope') # 2 re-nope cands conn.commit(); conn.close() recipe, items, _e = _v11_compose(db_path) slots = _slots(items) assert len(items) == 12 and len(slots['new']) == 0 \ and len(slots['return']) == 12 \ and len(slots['wildcard']) == 0, recipe n_nopes = sum(1 for it in slots['return'] if it['decision'] == 'nope') assert n_nopes <= 2, n_nopes first_ids = {it['id'] for it in items} # second compose on the same pool: LAST_TWO bars the 12 just served # and returns stay one-shot -> nothing to serve recipe2, items2, _e = _v11_compose(db_path) assert len(items2) == 0, items2 assert not ({it['id'] for it in items2} & first_ids) # --- A5: 40 fresh / 0 returns -> 25 then the 15 leftover, no # --- overlap, fresh ordered points DESC ---------------------------- conn = env() ids = _v11_ids(conn, 60) _v11_activate(conn, ids[0:40]) _v11_stage_fresh(conn, ids[0:40]) # spread scores so the points DESC order check is meaningful for k, lid in enumerate(ids[0:40]): conn.execute("UPDATE listings SET score=? WHERE id=?", (40 - k, lid)) conn.commit(); conn.close() recipe, items, engine = _v11_compose(db_path) assert len(items) == 25, len(items) assert recipe['return'] == 0, recipe first_ids = {it['id'] for it in items} slots1 = _slots(items) pts = [engine.points(it['id']) for it in slots1['new']] # scores unique DESC -> fresh-slot points non-increasing (wildcards # sit pinned at the back and don't participate in that order) assert pts == sorted(pts, reverse=True), pts recipe2, items2, _e = _v11_compose(db_path) slots2 = _slots(items2) assert recipe2['return'] == 0, recipe2 assert len(items2) == 15, len(items2) second_ids = {it['id'] for it in items2} assert not (first_ids & second_ids) # zero overlap # --- A6: 0/0/0 -> empty composed section (FE side = item 8) -------- conn = env() conn.commit(); conn.close() recipe, items, _e = _v11_compose(db_path) assert items == [], items assert (recipe['new'], recipe['return'], recipe['wildcard']) \ == (0, 0, 0), recipe # --- A7: family cap — fill-heavy cruiser pool: fresh/reserved hold # --- <=4, fill raises return slots to <=8 -------------------------- conn = env() ids = _v11_ids(conn, 60) _v11_activate(conn, ids[0:40], family='cruiser') _v11_stage_fresh(conn, ids[0:10], seen=True) # 10 fresh cruiser _v11_stage_returns(conn, ids[10:40], 'maybe') # 30 cruiser rets conn.commit(); conn.close() recipe, items, _e = _v11_compose(db_path) slots = _slots(items) fam = {} for it in items: fam[it['family']] = fam.get(it['family'], 0) + 1 assert fam.get('cruiser', 0) <= 8, fam assert len(slots['new']) == 2, recipe # fresh stops at cap 4 assert len(slots['return']) == 6, recipe # 2 reserved + 4 fill assert (recipe['new'], recipe['return']) == (2, 6), recipe # --- A8: re-nope gate — <=2 nope returns total; a rested item never # --- returns; gate can empty the reserved slots -------------------- conn = env() ids = _v11_ids(conn, 60) _v11_activate(conn, ids[0:10]) _v11_stage_returns(conn, ids[0:10], 'maybe') # 10 maybes _v11_activate(conn, ids[10:15], family='gatefam', score=100) _v11_stage_returns(conn, ids[10:15], 'nope') # 5 nope re-nopes # family net positive: gate passes for gatefam's nope candidates _v11_stage_support(conn, ids[30:40], family='gatefam') # 10 likes _v11_activate(conn, ids[15:16], family='gatefam') _v11_stage_returns(conn, ids[15:16], 'maybe', rests=[ids[15]]) # rested: r2 nope conn.commit(); conn.close() recipe, items, _e = _v11_compose(db_path) slots = _slots(items) assert len(items) == 12, len(items) # 10 maybe + 2 nope n_nopes = sum(1 for it in slots['return'] if it['decision'] == 'nope') assert n_nopes == 2, n_nopes assert all(it['id'] != ids[15] for it in slots['return']) # rested # gate-block: low-score nope family, net NOT positive -> reserved # slots skipped; fill still caps at 2 nopes (gate-free fill rule) conn = env() ids = _v11_ids(conn, 60) _v11_activate(conn, ids[0:3], family='blockfam', score=1) _v11_stage_returns(conn, ids[0:3], 'nope') # 3 gate-blocked conn.commit(); conn.close() recipe, items, _e = _v11_compose(db_path) slots = _slots(items) assert recipe['return'] == 2, recipe # fill cap only assert all(it['decision'] == 'nope' for it in slots['return']), items finally: shutil.rmtree(d, ignore_errors=True) P3_API_TESTS = [ ('p3: events seeding idempotent (from swipes)', test_p3_event_seeding_idempotent), ('p3: swipe round math + clear-then-re-decide', test_p3_round_math_and_clear), ('p3: section compose invariants (25/12, caps, bands)', test_p3_section_invariants), ('p3: exhaust -> section report', test_p3_exhaust_and_report), ('p3: forge modes (strict 409 · ondemand mock pass)', test_p3_forge_modes_strict_409), ('p3: trends endpoint shape + stats', test_p3_trends_shape_and_stats), ('p3: agentpass mock determinism + live fallback', test_p3_agentpass_mock_and_fallback), ('p3: points/band function known answers', test_p3_points_and_band), ('p3: payload additions (list/listing/section)', test_p3_payload_additions), ('p3: mid-section delist keeps the deck up (#3)', test_issue3_section_survives_midsection_delist), ] if API_DEPS: TESTS += P3_API_TESTS V11_TESTS = [ ('v1.1: item 5 — section counter keys off in-section events', test_v11_item5_section_counter), ('v1.1: item 7 — stats fresh = active minus swiped', test_v11_item7_stats_fresh), ('v1.1: item 8/6 — pool counts, 0/0/0 empty compose, dry forge note', test_v11_item8_pool_and_dry), ('v1.1: item 9 — batch formula v2 compose matrix A1-A8', test_v11_item9_compose_matrix), ] if API_DEPS: TESTS += V11_TESTS B1_TESTS = [ ('b1: guarded migrations (imgs + last_photo_idx, idempotent)', test_b1_migration_guarded), ('b1: importer carries imgs gallery (JSON, order, img fallback.)', test_b1_importer_gallery_carry), ('b1: 640-row copy re-import byte-stable (A)', test_b1_640row_reimport_idempotence), ] TESTS += B1_TESTS API_TESTS = [ ('api: / serves the app + static + manifest/sw', test_api_root), ('api: stats live == active count, health counts', test_api_stats_health), ('api: deck ordering + undecided-only + cursor pagination', test_api_deck), ('api: swipe upsert/edit/delete lifecycle', test_api_swipe_lifecycle), ('api: /img/{id} 200 cached | 302 hotlink | 404 none', test_api_img_routes), ('api: badge seen roundtrip', test_api_badge_roundtrip), ('b1: /img/{id}?i=n gallery chain (200/302/404, {id}_0 primary)', test_b1_img_route_chain), ('b1: swipe photo_idx persists / preserves / clears', test_b1_swipe_photo_idx), ('b1: payload n_photos + last_photo_idx, rotation math', test_b1_payload_fields_and_rotation_math), ] if API_DEPS: TESTS += API_TESTS def main(): failures = 0 for name, fn in TESTS: try: fn() print('PASS %s' % name) except Exception as e: failures += 1 print('FAIL %s: %s' % (name, e)) if not API_DEPS: print('NOTE P2 API checks skipped: fastapi/httpx not installed' ' (run via .venv/bin/python -m app.selfcheck)') print('---') print('%d/%d passed' % (len(TESTS) - failures, len(TESTS))) return 1 if failures else 0 if __name__ == '__main__': sys.exit(main())