MacKwnoX/Bike-Tinder
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

b1: selfcheck — 6 new B1 checks (guarded migration, imgs import + idempotence, 640-row copy byte-stable re-import, /img?i= chain, swipe photo_idx persistence, payload n_photos/last_photo_idx + rotation math); hardened two P3 checks against live-data drift on migrated copies (event seeding + profile sources); payload contract now includes B1 keys

Browse Source
This commit is contained in:
Torres
2026-11-11 00:10:56 -04:00
parent fff9f05bae
commit 38271d0d88
1 changed files with 393 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

408
app/selfcheck.py
View File

@@ -1,9 +1,12 @@
"""Dependency-free self-check for the Phase 1 data spine.
"""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, and gone-after-2-misses.
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
@@ -308,12 +311,23 @@ TESTS = [
PAYLOAD_KEYS = {'id', 'title', 'price_display', 'price_num', 'km', 'location',
'region', 'url', 'img', 'desc', 'family', 'score', 'year',
'seller_type', 'extras', 'decision', 'first_seen'}
'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.path.join(REPO_ROOT, 'data', 'bike-tinder.sqlite3')
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. ---------
@@ -552,20 +566,32 @@ def test_p3_event_seeding_idempotent():
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']
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']
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(
@@ -803,7 +829,10 @@ def test_p3_agentpass_mock_and_fallback():
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
# 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)
@@ -817,7 +846,8 @@ def test_p3_agentpass_mock_and_fallback():
ag.validate_payload(res3['payload'])
conn = _sq_conn(db_path)
sources = [r['source'] for r in conn.execute(
'SELECT source FROM taste_profile ORDER BY version')]
'SELECT source FROM taste_profile WHERE version >= ?'
' ORDER BY version', (new_min,))]
assert sources == ['mock', 'fallback'], sources
conn.close()
finally:
@@ -917,6 +947,338 @@ def test_p3_payload_additions():
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)
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),
@@ -933,6 +1295,16 @@ P3_API_TESTS = [
if API_DEPS:
TESTS += P3_API_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),
@@ -940,6 +1312,12 @@ API_TESTS = [
('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: