tricu

web.py (2726B)

---

 #!/usr/bin/env python3
 """Repeated requests in one PHP process exercise request-scoped FFI teardown."""
 import os
 from pathlib import Path
 import socket
 import subprocess
 import sys
 import tempfile
 import time
 import urllib.request
 import urllib.error
 
 php, public, fixture = sys.argv[1:]
 with socket.socket() as sock:
     sock.bind(('127.0.0.1',0))
     port=sock.getsockname()[1]
 with tempfile.TemporaryFile() as log:
     server=subprocess.Popen([php,'-d','ffi.enable=true','-S',f'127.0.0.1:{port}','-t',public],stdout=log,stderr=log,
                             env={k:v for k,v in os.environ.items() if k!='PHP_CLI_SERVER_WORKERS'})
     def post(bundle,name='id',format='string',argument='hello 🌳\x00world'):
         boundary='tricu-binding-test-boundary'
         fields={'export':name,'format':format,'arg0':argument}
         body=b''
         for key,value in fields.items():
             body+=f'--{boundary}\r\nContent-Disposition: form-data; name="{key}"\r\n\r\n'.encode()+value.encode()+b'\r\n'
         body+=f'--{boundary}\r\nContent-Disposition: form-data; name="bundle"; filename="test.arboricx"\r\nContent-Type: application/octet-stream\r\n\r\n'.encode()
         body+=bundle+f'\r\n--{boundary}--\r\n'.encode()
         request=urllib.request.Request(f'http://127.0.0.1:{port}/eval.php',body,
             {'Content-Type':f'multipart/form-data; boundary={boundary}'})
         try:
             with urllib.request.urlopen(request,timeout=5) as response:
                 return response.status,response.read()
         except urllib.error.HTTPError as error:
             return error.code,error.read()
     try:
         for _ in range(100):
             if server.poll() is not None: raise RuntimeError('PHP server exited during startup')
             try:
                 with socket.create_connection(('127.0.0.1',port),timeout=.1): break
             except OSError: time.sleep(.02)
         else: raise RuntimeError('PHP server startup timed out')
         bundle=Path(fixture).read_bytes()
         for i in range(12):
             assert post(bundle)==(200,'hello 🌳\x00world'.encode())
             assert post(bundle,name='missing')[0]==400
             assert post(b'bad bundle')[0]==400
             assert post(bundle,format='invalid')[0]==400
             assert server.poll() is None, 'PHP process died between requests'
         assert post(bundle)==(200,'hello 🌳\x00world'.encode())
         print('PHP request lifecycle passed in one persistent process')
     except BaseException:
         log.seek(0); sys.stderr.write(log.read().decode(errors='replace')); raise
     finally:
         server.terminate()
         try: server.wait(timeout=5)
         except subprocess.TimeoutExpired: server.kill(); server.wait()