MacKwnoX/Bike-Tinder
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity
Files
98d345ceee9545827220e0ecde3b0968ccca2896
Bike-Tinder/app/selfcheck.py
Torres 98d345ceee fix(v1.1): item 5 — section counter keys off in-section events (2/2 -> 3/2 bug) 
A card is decided only when a swipe event carries THIS section's id;
the lifetime swipes.action a second look wears into its section no longer
paints it as pre-decided. progress/undecided, report counts (latest
in-section event per listing) and maybe_exhaust all use the new
section_decided_ids() set; read path only, no writes. Every section_state
item carries decided_this_section; the FE counts server truth unless a
local intent is in flight, and sectDecided is clamped to sectTotal
everywhere (commit + paintSect) so the pill can never read past N/N.

Semantics note: completing a section now means striking EVERY item in it
(returns re-struck on their second look) — the exhaust selfcheck was
updated to strike all items, matching the deck's real flow. Undo-on-a-
return residual (event stays after the DELETE) is the accepted Q5 clamp-
only behavior, covered by a dedicated selfcheck case.
2026-11-13 13:38:18 -04:00

1437 lines
63 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""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', 'new', 'likes', 'maybes', 'nopes', 'last_import'}
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'
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)
# --- 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)
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),
]
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),
]
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())
Reference in New Issue View Git Blame Copy Permalink