MacKwnoX/Bike-Tinder
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity
Files
main
Bike-Tinder/app/selfcheck.py
Alexander Rozhenko beeace231d fix #3: guard breakdown() so a delisted mid-batch card cannot 500 the deck 
A section can outlive an ad: when a card's listing left the active pool after compose and has no swipe history, it is absent from the engine's by_id map. breakdown() indexed the missing row without a guard, so one such card 500'd GET /api/section (and /api/listing/{id}) - live incident 2026-11-15: section 57 + delisted listing 298. Degrade to no-signal lines instead. Regression test added; before/after verified on a copy of the live DB (section 500 -> 200; forge rebuild 200).
2026-11-15 11:06:51 -04:00

1861 lines
82 KiB
Python
Raw Permalink 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', '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())
Reference in New Issue View Git Blame Copy Permalink