| """Real localhost HTTP integration; server always stopped before this script exits.""" |
| from pathlib import Path |
| import sys,subprocess,tempfile,os,secrets,socket,time,json,urllib.request,urllib.error |
| ROOT=Path(__file__).resolve().parents[1] |
| def main(): |
| with tempfile.TemporaryDirectory() as td: |
| key=secrets.token_urlsafe(40);env=dict(os.environ,PHILLNET_RECALL_API_KEY=key,PYTHONPATH=str(ROOT)) |
| with socket.socket() as s:s.bind(('127.0.0.1',0));port=s.getsockname()[1] |
| db=Path(td)/'memory.db';base=f'http://127.0.0.1:{port}' |
| def request(path,data=None,auth=True): |
| headers={'Content-Type':'application/json'} |
| if auth:headers['Authorization']='Bearer '+key |
| r=urllib.request.Request(base+path,data=None if data is None else json.dumps(data).encode(),headers=headers) |
| with urllib.request.urlopen(r,timeout=5) as response:return response.status,json.load(response) |
| def start(): |
| proc=subprocess.Popen([sys.executable,'-m','phillnet_recall','--db',str(db),'serve','--port',str(port)],cwd=ROOT,env=env,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL) |
| for _ in range(100): |
| if proc.poll() is not None:raise RuntimeError('server failed startup') |
| try:request('/health',auth=False);return proc |
| except (urllib.error.URLError,TimeoutError):time.sleep(.05) |
| proc.terminate();proc.wait(5);raise RuntimeError('server startup timed out') |
| proc=start() |
| try: |
| try:request('/v1/stats',auth=False);raise AssertionError('unauthenticated request allowed') |
| except urllib.error.HTTPError as e:assert e.code==401 |
| status,data=request('/v1/memories',{'text':'Willow uses port 7442.','source':'synthetic-smoke'}) |
| ident=data['id'];assert status==200 |
| status,data=request('/v1/search',{'query':'Willow port'});assert data['items'][0]['id']==ident |
| _,x=request('/v1/memories',{'text':'Willow uses Beacon','source':'synthetic','entity':'Willow','predicate':'uses_service','value':'Beacon'}) |
| _,y=request('/v1/memories',{'text':'Beacon region north','source':'synthetic','entity':'Beacon','predicate':'deployment_region','value':'north'}) |
| _,path=request('/v1/recall',{'query':'Which region hosts the service used by Willow?'}) |
| assert path['answer_values']==['north'] |
| _,p=request('/v1/evidence-context',{'query':'Which region hosts the service used by Willow?','budget':8192}) |
| assert p['status']=='ready' and p['dependency_complete'] |
| finally:proc.terminate();proc.wait(10) |
| proc=start() |
| try: |
| _,data=request('/v1/search',{'query':'Willow port'});assert data['items'][0]['id']==ident |
| _,packet=request('/v1/context',{'query':'Willow port','budget':1024});assert packet['used']<=1024 |
| _,path=request('/v1/paths',{'entity':'Willow','steps':[{'predicate':'uses_service'},{'predicate':'region'}]}) |
| assert path['answer_values']==['north'] |
| finally:proc.terminate();proc.wait(10) |
| report={'transport':'real localhost HTTP','unauthorized_status':401,'write':True,'retrieve':True,'restart_persistence':True,'context_budget':True,'multi_hop_recall':True,'proof_packet_complete':True,'path_after_restart':True,'server_left_running':False,'docker_built':False} |
| (ROOT/'reports/http_smoke.json').write_text(json.dumps(report,indent=2)+'\n');print(json.dumps(report,indent=2)) |
| if __name__=='__main__':main() |
|
|
