250 lines
10 KiB
Python
250 lines
10 KiB
Python
"""Dependency-free self-check for the Phase 1 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.
|
|
"""
|
|
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)
|
|
|
|
|
|
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),
|
|
('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),
|
|
]
|
|
|
|
|
|
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))
|
|
print('---')
|
|
print('%d/%d passed' % (len(TESTS) - failures, len(TESTS)))
|
|
return 1 if failures else 0
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|