import json, urllib.request, urllib.error, sys, datetime B='http://127.0.0.1:8080/api/v1' res=[] def call(method, path, body=None, token=None, headers=None): h={'Accept':'application/json'} if body is not None: h['Content-Type']='application/json' if token: h['Authorization']='Bearer '+token if headers: h.update(headers) req=urllib.request.Request(B+path, data=json.dumps(body).encode() if body is not None else None, method=method, headers=h) try: with urllib.request.urlopen(req) as r: return r.status, json.loads(r.read() or b'{}'), dict(r.headers) except urllib.error.HTTPError as e: raw=e.read() try: return e.code, json.loads(raw), dict(e.headers) except Exception: return e.code, {'raw':raw[:300].decode('utf8','replace')}, dict(e.headers) def check(name, cond, extra=''): res.append(bool(cond)); print(('PASS ' if cond else 'FAIL ')+name+('' if cond else ' -> '+str(extra)[:400])) # --- auth --- s,j,_=call('POST','/auth/register',{'email':'kid@example.com','password':'longenough1','date_of_birth':'2015-01-01','accepted_terms':True}) check('register under-18 rejected (403)', s==403, (s,j)) s,j,_=call('POST','/auth/register',{'email':'bad','password':'x','date_of_birth':'2000-01-01'}) check('register validation (422)', s==422 and 'details' in j.get('error',{}), (s,j)) s,j,_=call('POST','/auth/register',{'email':'tester@example.com','password':'longenough1','date_of_birth':'1990-05-05','accepted_terms':True,'display_name':'Tester'}) check('register adult (201)', s==201 and 'tokens' in j.get('data',{}), (s,j)) tok=j['data']['tokens']['access_token']; ref=j['data']['tokens']['refresh_token'] s,j,_=call('POST','/auth/register',{'email':'tester@example.com','password':'longenough1','date_of_birth':'1990-05-05','accepted_terms':True}) check('duplicate email rejected (422)', s==422, (s,j)) s,j,_=call('POST','/auth/login',{'email':'tester@example.com','password':'wrongwrong1'}) check('bad password 401', s==401, (s,j)) s,j,_=call('POST','/auth/login',{'email':'tester@example.com','password':'longenough1'}) check('login ok', s==200 and j['data']['user']['role']=='viewer', (s,j)) tok=j['data']['tokens']['access_token']; ref=j['data']['tokens']['refresh_token'] s,j,_=call('GET','/auth/me',token=tok); check('me', s==200 and j['data']['email']=='tester@example.com', (s,j)) s,j,_=call('GET','/auth/me',token='0'*64); check('invalid token 401', s==401, (s,j)) s,j,_=call('POST','/auth/refresh',{'refresh_token':ref}); check('refresh rotates', s==200 and j['data']['tokens']['access_token']!=tok, (s,j)) new=j['data']['tokens'] s,j,_=call('POST','/auth/refresh',{'refresh_token':ref}); check('old refresh reuse rejected', s==401, (s,j)) s,j,_=call('GET','/auth/me',token=new['access_token']); check('reuse revoked all sessions', s==401, (s,j)) s,j,_=call('POST','/auth/login',{'email':'tester@example.com','password':'longenough1'}); tok=j['data']['tokens']['access_token'] # --- data --- today=datetime.datetime.now(datetime.UTC).replace(tzinfo=None) s,j,_=call('GET','/competitions',token=tok); check('competitions', s==200 and j['data'][0]['name']=='Test Premier League', (s,j)) cid=j['data'][0]['id'] s,j,_=call('GET','/seasons?competition_id=%d'%cid,token=tok); check('seasons', s==200 and len(j['data'])==1, (s,j)) s,j,_=call('GET','/seasons',token=tok); check('seasons requires competition_id (422)', s==422, (s,j)) frm=(today-datetime.timedelta(days=27)).strftime('%Y-%m-%d'); to=(today+datetime.timedelta(days=3)).strftime('%Y-%m-%d') s,j,_=call('GET','/fixtures?from=%s&to=%s&per_page=100'%(frm,to),token=tok); check('fixtures range', s==200 and j['meta']['total']>0, (s,j)) s,j,_=call('GET','/fixtures?from=2026-01-01&to=2026-12-31',token=tok); check('range >31d rejected', s==422, (s,j)) s,j,_=call('GET','/fixtures?date=2026-02-31',token=tok); check('invalid date rejected', s==422, (s,j)) s,j,_=call("GET","/fixtures?date=2026-10-08%27%20OR%201%3D1%20--",token=tok); check('SQLi in date rejected', s==422, (s,j)) s,j,_=call('GET','/fixtures?status=bogus&date=2026-10-08',token=tok); check('bad status rejected', s==422, (s,j)) s,j,_=call('GET','/fixtures?status=scheduled&from=%s&to=%s'%(today.strftime('%Y-%m-%d'),(today+datetime.timedelta(days=7)).strftime('%Y-%m-%d')),token=tok) check('scheduled fixtures listed', s==200 and j['meta']['total']>=1, (s,j)) up=j['data'][0]; fid=up['id'] s,j,_=call('GET','/fixtures/%d'%fid,token=tok); check('fixture detail + data_state', s==200 and j['data']['data_state']['predictions']=='available', (s,j)) s,j,_=call('GET','/fixtures/999999',token=tok); check('fixture 404', s==404, (s,j)) s,j,_=call('GET','/fixtures/%d/stats'%fid,token=tok); check('stats missing state for scheduled', s==200 and j['data']['state']=='missing', (s,j)) # finished fixture stats s,j,_=call('GET','/fixtures?status=finished&from=%s&to=%s&per_page=5'%(frm,to),token=tok) fin=j['data'][0]['id'] if j['data'] else None s,j,_=call('GET','/fixtures/%d/stats'%fin,token=tok); t=j['data']['teams'][0] if j['data']['state']=='available' else {} check('stats available with null for missing', s==200 and t.get('fouls',1) is None and t.get('shots') is not None, (s,j)) # --- predictions --- s,j,_=call('GET','/fixtures/%d/predictions'%fid,token=tok); d=j.get('data',{}) m=d.get('markets',{}) ok=s==200 and d.get('state')=='available' ssum=sum(m['1X2'][k]['probability'] for k in ('HOME','DRAW','AWAY')) if ok else 0 check('prediction available, 1X2 sums ~1', ok and abs(ssum-1)<0.001, (s,j)) check('prediction has OU lines, DC, BTTS, scorelines', ok and '2.5' in m['OU'] and 'DC' in m and 'BTTS' in m and len(d['top_scorelines'])==6, d.keys() if ok else j) check('DC 1X = HOME+DRAW', ok and abs(m['DC']['1X']['probability']-(m['1X2']['HOME']['probability']+m['1X2']['DRAW']['probability']))<0.0002) check('prediction exposes quality + pre_match + model', ok and d['run']['pre_match'] is True and d['data_quality']['level'] in ('low','medium','high'), d.get('run') if ok else j) s,j,_=call('GET','/models/performance',token=tok); check('performance empty state honest', s==200 and j['data']['sample_size']==0, (s,j)) s,j,_=call('GET','/predictions/history',token=tok); check('history empty', s==200 and j['meta']['total']==0, (s,j)) # --- odds & insights --- s,j,_=call('GET','/fixtures/%d/odds'%fid,token=tok); d=j.get('data',{}) check('odds available w/ 3 books', s==200 and d.get('state')=='available' and len(d['bookmakers'])==3, (s,j)) if d.get('state')=='available': sel={x['key']:x for x in d['selections']} check('best odds >= every book (spot check HOME)', sel['1X2:HOME']['best_odds']>1 and sel['1X2:HOME']['bookmakers_quoting']==3) nv=sum(sel['1X2:'+k]['market_probability'] for k in ('HOME','DRAW','AWAY')) check('market probs sum to 1', abs(nv-1)<0.002, nv) check('overround positive ~5-8%', all(0.03=1 and 'note' in j['meta'], (s,j)) s,j,_=call('GET','/scanner',token=tok); check('scanner default status=value is empty while unvalidated', s==200 and j['meta']['total']==0, (s,j)) s,j,_=call('GET','/scanner?market=ZZ',token=tok); check('scanner bad market 422', s==422, (s,j)) s,j,_=call('GET','/scanner?min_edge=7',token=tok); check('scanner bad min_edge 422', s==422, (s,j)) s,j,_=call('GET','/scanner?status=all&days=3&market=OU&min_probability=0.3',token=tok); check('scanner market filter', s==200 and all(r['key'].startswith('OU:') for r in j['data']), (s,j)) # --- watchlists / alerts --- s,j,_=call('POST','/watchlists',{'name':'Mine'},token=tok); check('watchlist create', s==201, (s,j)); wid=j['data']['id'] s,j,_=call('POST','/watchlists',{'name':'Mine'},token=tok); check('duplicate watchlist 409', s==409, (s,j)) s,j,_=call('POST','/watchlists/%d/fixtures'%wid,{'fixture_id':fid},token=tok); check('add fixture', s==201, (s,j)) s,j,_=call('POST','/watchlists/%d/fixtures'%wid,{'fixture_id':fid},token=tok); check('add fixture idempotent', s==201, (s,j)) s,j,_=call('GET','/watchlists',token=tok); check('watchlist list has fixture once', s==200 and len(j['data'][0]['fixtures'])==1, (s,j)) s,j,_=call('POST','/alerts',{'alert_type':'kickoff','fixture_id':fid},token=tok); check('alert create', s==201, (s,j)); aid=j['data']['id'] s,j,_=call('POST','/alerts',{'alert_type':'kickoff'},token=tok); check('alert needs fixture (422)', s==422, (s,j)) s,j,_=call('POST','/alerts',{'alert_type':'nope'},token=tok); check('alert bad type (422)', s==422, (s,j)) # second user cannot touch first user's data s,j,_=call('POST','/auth/register',{'email':'other@example.com','password':'longenough1','date_of_birth':'1991-01-01','accepted_terms':True}); t2=j['data']['tokens']['access_token'] s,j,_=call('DELETE','/watchlists/%d'%wid,token=t2); check("cannot delete another user's watchlist (404)", s==404, (s,j)) s,j,_=call('DELETE','/alerts/%d'%aid,token=t2); check("cannot delete another user's alert (404)", s==404, (s,j)) # --- RBAC --- s,j,_=call('GET','/admin/providers',token=tok); check('viewer blocked from admin (403)', s==403, (s,j)) s,j,_=call('POST','/admin/ingestion/jobs',{'job_type':'fixture_odds','params':{'fixture_id':1}},token=tok); check('viewer cannot enqueue (403)', s==403, (s,j)) print('\n%d/%d passed'%(sum(res),len(res))) open('/tmp/tokens.json','w').write(json.dumps({'tok':tok,'fid':fid,'fin':fin}))