review/test_api.py

"""Black-box adversarial checks against the local prototype only; no third-party URLs."""
from pathlib import Path
from urllib.request import Request,urlopen
from urllib.error import HTTPError
from copy import deepcopy
import json,time,hashlib,os
ROOT=Path(__file__).resolve().parent
BASE=os.environ.get('REVIEW_BASE_URL','http://127.0.0.1:4318')
results=[]

def post(value=None,raw=None,path='/api/triage',headers=None):
 data=raw.encode() if raw is not None else json.dumps(value,ensure_ascii=False).encode()
 request=Request(BASE+path,data=data,headers={'Content-Type':'application/json',**(headers or {})},method='POST')
 start=time.perf_counter()
 try:
  with urlopen(request,timeout=120) as response:status=response.status;body=response.read();hdrs=dict(response.headers)
 except HTTPError as response:status=response.code;body=response.read();hdrs=dict(response.headers)
 return {'status':status,'body':json.loads(body),'durationMs':(time.perf_counter()-start)*1000,'headers':hdrs}

def record(name,observed,passed,requirement,**detail):
 results.append({'name':name,'passed':bool(passed),'requirement':requirement,'response':observed,**detail})
 print(name,'PASS' if passed else 'FAIL',observed['status'])

cases=json.loads((ROOT/'adversarial_cases.json').read_text())['cases']
reject_names={'malformed_json','top_level_null','top_level_scalar','top_level_string','top_level_object','empty_list','non_object_item','missing_text','null_text','wrong_title_type','wrong_body_type','duplicate_id','duplicate_number','mixed_identifier','nonfinite_json'}
for case in cases:
 value=deepcopy(case.get('value'));original=deepcopy(value)
 # The declared app contract requires positive numeric IDs. Keep attack text and duplicate relationships intact.
 if isinstance(value,list):
  identity={}
  for ix,item in enumerate(value,1):
   if isinstance(item,dict) and 'id' in item and case['name'] not in {'mixed_identifier'}:
    key=str(item['id']);identity.setdefault(key,ix);item['id']=identity[key]
 response=post(value={'issues':value} if isinstance(value,list) else value,raw=case.get('raw'))
 if case['name'] in reject_names:passed=response['status']==400 and isinstance(response['body'].get('error'),str);requirement='Deliberate legible rejection under the documented schema.'
 else:passed=response['status']==200 and isinstance(response['body'].get('items'),list);requirement='Accept inert issue text; browser/export execution is checked separately.'
 if case['name']=='extra_fields':
  output=response['body'].get('items',[{}])[0];passed=passed and all(k not in output for k in ['__proto__','constructor','url','token'])
  requirement='Accept permitted issue fields and omit untrusted extra fields from emitted items.'
 record(case['name'],response,passed,requirement,adaptation='String fixture IDs mapped consistently to positive integers except the explicit mixed-type test.')

valid={'id':1,'title':'How do I find the logs?','body':'Which folder contains logs?','labels':[]}
extra=[
 ('title_over_300',{'issues':[{**valid,'title':'a'*301}]}),
 ('body_over_12000',{'issues':[{**valid,'body':'a'*12001}]}),
 ('items_over_100',{'issues':[{**valid,'id':i+1} for i in range(101)]}),
 ('total_normalized_over_256000',{'issues':[{**valid,'id':i+1,'body':'a'*11999} for i in range(22)]}),
 ('request_bytes_over_512000',{'issues':[{**valid,'id':i+1,'body':'é'*11000} for i in range(24)]}),
 ('labels_over_30',{'issues':[{**valid,'labels':['x'+str(i) for i in range(31)]}]}),
 ('label_object_without_name',{'issues':[{**valid,'labels':[{}]}]}),
 ('negative_id',{'issues':[{**valid,'id':-1}]}),
 ('unsafe_integer_id',{'issues':[{**valid,'id':9007199254740992}]}),
]
for name,payload in extra:
 response=post(payload);record(name,response,response['status']==400,'Bounded, legible input rejection.')

# Full boundary sizes are schema-only checks to avoid spending a large batch on repeated padding.
# Verify labels do not participate in model inference or become a source of expected truth.
a=post({'issues':[{**valid,'labels':[]}]})
b=post({'issues':[{**valid,'labels':['bug','<svg onload="globalThis.__triage_xss=7">']}]})
keys=['suggestedLabel','similarity','margin','needsReview','evidence']
record('labels_do_not_change_inference',b,a['status']==b['status']==200 and all(a['body']['items'][0][k]==b['body']['items'][0][k] for k in keys),'Existing labels are metadata and must not influence the predicted category.')

# Validate local apply and undo with a fresh run; no external service and no file mutation.
sample=ROOT.parent/'prototype/data/sample.json';before_hash=hashlib.sha256(sample.read_bytes()).hexdigest()
initial=post({'issues':[{**valid,'labels':['existing']},{'id':2,'title':'Please add an offline cache','body':'I request a new cache option.','labels':[]}]})
run=initial['body']['runId']
apply=post({'runId':run,'decisions':[{'id':1,'label':'question'}]},path='/api/apply')
passed=apply['status']==200 and apply['body']['issues'][0]['labels']==['existing','question'] and apply['body']['issues'][1]['labels']==[]
record('apply_selected_local_only',apply,passed,'Only the explicitly selected local issue gains the chosen label.')
undo=post({'revisionId':apply['body']['revisionId']},path='/api/undo')
record('undo_restores_exact_labels',undo,undo['status']==200 and undo['body']['issues'][0]['labels']==['existing'] and undo['body']['issues'][1]['labels']==[],'Undo restores the prior local working copy.')
repeat=post({'revisionId':apply['body']['revisionId']},path='/api/undo')
record('used_revision_rejected',repeat,repeat['status']==400,'An already consumed undo revision is rejected.')
stale=post({'runId':'nonexistent-run','decisions':[{'id':1,'label':'bug'}]},path='/api/apply')
record('stale_run_rejected',stale,stale['status']==400,'A stale run cannot apply changes.')
bad=post({'runId':run,'decisions':[{'id':1,'label':'bug'},{'id':1,'label':'feature'}]},path='/api/apply')
record('duplicate_decisions_rejected',bad,bad['status']==400,'Conflicting duplicate decisions are rejected.')
conflict_a=post({'runId':run,'decisions':[{'id':1,'label':'bug'}]},path='/api/apply')
conflict_b=post({'runId':run,'decisions':[{'id':2,'label':'feature'}]},path='/api/apply')
conflict=post({'revisionId':conflict_a['body']['revisionId']},path='/api/undo')
record('older_undo_conflict_rejected',conflict,conflict['status']==400,'Older undo does not clobber subsequent local changes.')
source_unchanged=hashlib.sha256(sample.read_bytes()).hexdigest()==before_hash
summary={'synthetic':True,'baseUrl':BASE,'count':len(results),'passed':sum(r['passed'] for r in results),'failed':[r['name'] for r in results if not r['passed']],'originalSampleFileUnchanged':source_unchanged,'results':results}
(ROOT/'api_results.json').write_text(json.dumps(summary,ensure_ascii=False,indent=2)+'\n')
print(json.dumps({k:v for k,v in summary.items() if k!='results'},indent=2))