tricu

io_v2_clock.py (3267B)

---

 #!/usr/bin/env python3
 """Actual source timing, scoped waits/cancellation, and migrated async demos."""
 import argparse
 import json
 from pathlib import Path
 import subprocess
 from source_order import Runner
 from http_headers import number, result, tree_list
 from io_v2_objects import text
 from io_v2_directory import parse_tree, natural
 
 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();root=a.directory.resolve();root.mkdir(parents=True,exist_ok=False)
     runner=Runner(a.tricu,root);checks=[]
     def compile(label,path):
         bundle=root/(label+'.arboricx')
         runner.run(label+'-compile',['compile','--file',path,'--store',runner.store,'--output',bundle,'--compile-timeout','600'],seconds=660)
         return runner.imported(bundle,label+'-import')
     pin=compile('clock',ROOT/'demos/io-v2/clock.tri')
     def invoke(label,name,value=None):
         args=['--io']
         if value is not None:
             path=root/(label+'.tree');path.write_bytes(value);args+=['--arg-tree-file',path]
         return runner.invoke(label,pin,name,args)
     assert invoke('yield','yielded')==result(True,b'\0');checks.append('yield-unit')
     assert invoke('cancelled','cancelledWait')==tree_list([text('cancelled'),text('timer-stop'),b'\0']);checks.append('cancel-max-duration')
     for milliseconds in [0,1,20]:
         raw=invoke(f'elapsed-{milliseconds}','elapsed',number(milliseconds))
         # Decode the binary kernel tree with the existing reader; compare to an
         # independent host duration bound, never a self-derived expected value.
         elapsed=natural(parse_tree(raw))
         assert elapsed>=milliseconds*1000000,(milliseconds,elapsed)
         checks.append(f'elapsed-{milliseconds}')
     assert natural(parse_tree(invoke('scope-wait','scopeWait')))>=20000000
     checks.append('normal-scope-waits')
     problem=tree_list([text('invalid-argument'),text('clock/sleep'),b'\0',text('none')])
     for i,value in enumerate([number(9223372036854775808),b'\1\0',b'\2\0\0']):
         assert invoke(f'invalid-{i}','invalidSleep',value)==result(False,problem)
         checks.append(f'invalid-{i}')
     for label,path,wanted in [
       ('async',ROOT/'demos/interactionTrees.tri',b'Parent first!\n2000ms done sleeping!\n5000ms done sleeping!\n'),
       ('yield-demo',ROOT/'demos/interactionTrees/yield.tri',b'A says 1\nA says 2\nA says 3\nB says 1\nB says 2\nB says 3\nFinished: doneA doneB\n')]:
         demo=compile(label,path)
         command=[str(runner.exe),'arboricx','run','--module',demo,'--name','main','--store',str(runner.store),'--io','--no-result']
         run=subprocess.run(command,stdout=subprocess.PIPE,stderr=subprocess.PIPE,timeout=30)
         (root/(label+'.stdout')).write_bytes(run.stdout);(root/(label+'.stderr')).write_bytes(run.stderr)
         assert run.returncode==0,(label,run.stderr)
         assert run.stdout==wanted,(label,run.stdout)
         checks.append(label)
     (root/'report.json').write_text(json.dumps(dict(passed=checks,records=runner.records),indent=2)+'\n')
     print(f'PASS {len(checks)} clock/caller controls')
 if __name__=='__main__':main()