tricu

io_process.py (4081B)

---

 #!/usr/bin/env python3
 """Independent actual-command, binary pipeline and normal-scope-exit controls."""
 import argparse
 import json
 import shutil
 from pathlib import Path
 from source_order import Runner
 from http_headers import number, tree_list, result
 from io_objects 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(); directory=a.directory.resolve(); directory.mkdir(parents=True,exist_ok=False)
     runner=Runner(a.tricu,directory); bundle=directory/'process.arboricx'; checks=[]
     runner.run('compile',['compile','--file',ROOT/'demos/io/process.tri','--store',runner.store,'--output',bundle,'--compile-timeout','600'],seconds=660)
     pin=runner.imported(bundle,'import')
     inherit=tree_list([text('inherit')])
     def invoke(label,name,args):
         path=directory/(label+'.tree');path.write_bytes(tree_list(args))
         return runner.invoke(label,pin,name,['--io','--arg-tree-file',path])
     cat=shutil.which('cat');shell=shutil.which('sh');assert cat and shell
     source=directory/'input';output=directory/'output';errors=directory/'errors'
     def pipeline(label,exe,argv,cwd,env,wanted,err=b'',code=0):
         actual=invoke(label,'pipeline',[text(exe),tree_list([text(x) for x in argv]),cwd,env,number(20),text(str(source)),text(str(output)),text(str(errors))])
         assert actual==tree_list([text('exit'),number(code)]),(label,actual.hex())
         assert output.read_bytes()==wanted,label
         assert errors.read_bytes()==err,label
         checks.append(label)
     for size in [0,1,4095,4096,4097,65536,1048576]:
         data=bytes(i%256 for i in range(size));source.write_bytes(data)
         pipeline(f'binary-{size}',cat,[],inherit,inherit,data)
     tee=shutil.which('tee');assert tee
     data=bytes(range(256))*4096;source.write_bytes(data)
     pipeline('concurrent-stdout-stderr',tee,['/dev/stderr'],inherit,inherit,data,data)
     source.write_bytes(b'')
     pipeline('literal-argv-env-cwd',shell,['-c','printf "%s|%s|%s" "$1" "$VALUE" "$PWD"; printf diagnostic >&2; exit 7','test','$(not-a-command); *'],tree_list([text('at'),text(str(directory))]),tree_list([text('replace'),tree_list([tree_list([text('VALUE'),text('literal')])])]),('$(not-a-command); *|literal|'+str(directory)).encode(),b'diagnostic',7)
     marker=directory/'joined'
     actual=invoke('normal-exit','normalExit',[text(shell),tree_list([text('-c'),text('i=0; while [ "$i" -lt 10000 ]; do i=$((i+1)); done; printf finished > "$1"; exit 3'),text('test'),text(str(marker))])])
     assert actual==number(42) and marker.read_bytes()==b'finished'
     checks.append('normal-exit-reaps-without-implying-zero-status')
     for label,exe,args,cwd,env,grace in [
         ('empty-executable','',[],inherit,inherit,0),
         ('duplicate-environment',cat,[],inherit,tree_list([text('replace'),tree_list([tree_list([text('X'),text('a')]),tree_list([text('X'),text('b')])])]),0),
         ('excess-argv',cat,['a']*257,inherit,inherit,0),
         ('excess-text',cat,['a'*4097],inherit,inherit,0),
         ('excess-total',cat,['a'*4096]*16,inherit,inherit,0),
         ('excess-grace',cat,[],inherit,inherit,9223372036854775808)]:
         actual=invoke(label,'raw',[text(exe),tree_list([text(x) for x in args]),cwd,env,number(grace)])
         assert actual==result(False,tree_list([text('invalid-argument'),text('process/spawn'),b'\0',text('none')])),label
         checks.append(label)
     for row in runner.records:
         if row.get('label') in ('binary-1048576','concurrent-stdout-stderr'):
             # Runtime process observations, not subprocess heap or capacity promises.
             assert row['rts_process']['maximum_live_bytes'] < 16*1024*1024,row
             assert row['peak_kib'] < 128*1024,row
     (directory/'report.json').write_text(json.dumps(dict(passed=checks,records=runner.records),indent=2)+'\n')
     print(f'PASS {len(checks)} process source controls')
 if __name__=='__main__':main()