File size: 3,559 Bytes
3b3b2bd | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 | """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()
|