tricu

io_v2_demo_servers.py (5515B)

---

 #!/usr/bin/env python3
 """Actual migrated TCP/HTTP/registry demo lifetimes and independent wire bytes."""
 import argparse,json,socket,subprocess,threading,hashlib,selectors
 from pathlib import Path
 from source_order import Runner
 from http_headers import number,tree_list
 from io_v2_text import text
 ROOT=Path(__file__).resolve().parent.parent
 
 def main():
     p=argparse.ArgumentParser(description=__doc__);p.add_argument('tricu',type=Path);p.add_argument('--directory',type=Path,required=True)
     a=p.parse_args();d=a.directory.resolve();d.mkdir(parents=True,exist_ok=False);r=Runner(a.tricu,d);checks=[];pins={}
     def build(name):
         bundle=d/(name+'.arboricx');r.run(name+'-compile',['compile','--file',ROOT/f'demos/interactionTrees/{name}.tri','--store',r.store,'--output',bundle,'--compile-timeout','600'],seconds=660)
         pins[name]=r.imported(bundle,name+'-import');return pins[name]
     def command(name,pin,values):
         command=[str(r.exe),'arboricx','run','--module',pin,'--name','server','--store',str(r.store),'--io','--no-result']
         for i,value in enumerate(values):
             f=d/f'{name}-{i}.arg';f.write_bytes(value);command+=['--arg-tree-file',str(f)]
         return command
     def start(name,pin,values):
         proc=subprocess.Popen(command(name,pin,values),stdout=subprocess.PIPE,stderr=subprocess.PIPE)
         with selectors.DefaultSelector() as sel:
             sel.register(proc.stdout,selectors.EVENT_READ);assert sel.select(15),'no listen announcement'
         port=int(proc.stdout.readline());return proc,port
     def close(proc,port):
         out,err=proc.communicate(timeout=20);assert proc.returncode==0,(out,err)
         try: socket.create_connection(('127.0.0.1',port),timeout=1).close()
         except OSError: return
         raise AssertionError('listener outlived owning scope')
     def endpoint(port=0):return tree_list([text('ipv4'),tree_list([number(x) for x in [127,0,0,1]]),number(port)])
     def request(port,data):
         with socket.create_connection(('127.0.0.1',port),timeout=20) as s:
             s.sendall(data);s.shutdown(socket.SHUT_WR);parts=[]
             while True:
                 b=s.recv(4096)
                 if not b:return b''.join(parts)
                 parts.append(b)
     proc,port=start('echo',build('echoServer'),[endpoint(),number(2)])
     try:
         for data in [b'',bytes(range(256))*1024]:
             with socket.create_connection(('127.0.0.1',port),timeout=20) as s:
                 failures=[]
                 def send():
                     try:s.sendall(data);s.shutdown(socket.SHUT_WR)
                     except Exception as e:failures.append(e)
                 thread=threading.Thread(target=send);thread.start();out=[]
                 while True:
                     chunk=s.recv(4096)
                     if not chunk:break
                     out.append(chunk)
                 thread.join(5);assert not thread.is_alive() and not failures
                 assert b''.join(out)==data
         close(proc,port);checks.append('echo-binary-and-scope-close')
     finally:
         if proc.poll() is None:proc.kill();proc.wait()
     cases=[(b'GET /hello HTTP/1.1\r\nHost: test\r\n\r\n',200,b'Hello from /hello\n'),(b'HEAD /hello HTTP/1.1\r\nHost: test\r\n\r\n',200,b''),(b'POST /hello HTTP/1.1\r\nHost: test\r\nContent-Length: 0\r\n\r\n',405,b'Method not allowed\n'),(b'bad\r\n\r\n',400,b'Bad Request\n')]
     proc,port=start('http',build('httpServer'),[endpoint(),number(len(cases))])
     try:
         for payload,status,body in cases:
             reply=request(port,payload);head,actual=reply.split(b'\r\n\r\n',1)
             assert head.startswith(f'HTTP/1.1 {status} '.encode()) and actual==body,(head,actual)
             length=len(b'Hello from /hello\n') if payload.startswith(b'HEAD') else len(body)
             assert f'Content-Length: {length}\r\n'.encode() in head+b'\r\n'
         close(proc,port);checks.append('http-methods-framing-head-and-scope-close')
     finally:
         if proc.poll() is None:proc.kill();proc.wait()
     root=d/'registry';data=b'\0\xffregistry-demo';digest=hashlib.sha256(data).hexdigest()
     proc,port=start('registry',build('arboricxServer'),[text(str(root)),endpoint(),number(3)])
     try:
         assert request(port,b'GET /_arboricx/health HTTP/1.1\r\nHost: test\r\n\r\n').endswith(b'OK\n')
         response=request(port,b'POST /_arboricx/bundle HTTP/1.1\r\nHost: test\r\nContent-Length: '+str(len(data)).encode()+b'\r\n\r\n'+data)
         assert response.startswith(b'HTTP/1.1 201 ') and response.endswith(digest.encode()),response
         assert request(port,f'GET /_arboricx/bundle/hash/{digest} HTTP/1.1\r\nHost: test\r\n\r\n'.encode()).endswith(b'\r\n\r\n'+data)
         close(proc,port);checks.append('registry-binary-upload-download-and-scope-close')
     finally:
         if proc.poll() is None:proc.kill();proc.wait()
     for name,values in [('echoServer',[endpoint(),number(0)]),('httpServer',[endpoint(),number(0)]),('arboricxServer',[text(str(d/'failed-ready-store')),endpoint(),number(0)])]:
         with open('/dev/full','wb') as full:
             failed=subprocess.run(command(name+'-failed-ready',pins[name],values),stdout=full,stderr=subprocess.PIPE,timeout=20)
         assert failed.returncode!=0,(name,failed.stderr)
         checks.append(name+'-announcement-failure')
     (d/'report.json').write_text(json.dumps(dict(passed=checks,records=r.records),indent=2)+'\n');print('PASS migrated server demo wire/lifecycle controls')
 if __name__=='__main__':main()