tricu

io_v2_zig_stages.py (3731B)

---

 #!/usr/bin/env python3
 """Cross-directory publication, EXDEV, invalid destinations and no rollback."""
 import argparse
 import hashlib
 import json
 import subprocess
 import tempfile
 from pathlib import Path
 from source_order import Runner
 from http_headers import result,tree_list,number
 from process_rss import run_guarded
 ROOT=Path(__file__).resolve().parent.parent
 
 def text(s):return tree_list([number(ord(c)) for c in str(s)])
 def main():
     p=argparse.ArgumentParser(description=__doc__);p.add_argument('tricu',type=Path);p.add_argument('probe',type=Path);p.add_argument('--directory',type=Path,required=True)
     args=p.parse_args();d=args.directory.resolve();d.mkdir(parents=True,exist_ok=False);r=Runner(args.tricu,d);bundle=d/'stages.arboricx'
     r.run('compile',['compile','--file',ROOT/'test/io-v2/zig-stages.tri','--store',r.store,'--output',bundle]);pin=r.imported(bundle,'import');checks=[]
     a,b=d/'a',d/'b';a.mkdir();b.mkdir();target=b/'published'
     def invoke(name,values,label,expected):
         command=[str(args.probe.resolve()),str(bundle),name,*[str(v) if isinstance(v,int) else 'text:'+str(v) for v in values]]
         output,peak=run_guarded(command,memory_mib=128,seconds=60);(d/(label+'.log')).write_text(output)
         assert bytes(map(int,output.strip()))==expected,label
         checks.append(dict(label=label,peak_kib=peak))
     for mode,state in [(0,'published'),(0,'exists'),(1,'published')]:
         invoke('publish',[a,target,mode],f'publish-{mode}-{state}',result(True,text(state)))
         assert target.read_bytes()==b'abc' and list(a.iterdir())==[]
     target.unlink()  # prove visibility was created before the terminal failure
     process=subprocess.run([str(args.probe.resolve()),str(bundle),'failureAfterPublication','text:'+str(a),'text:'+str(target)],capture_output=True,timeout=60)
     assert process.returncode==1 and b'failed (payload not forced)' in process.stderr
     assert target.read_bytes()==b'abc' and list(a.iterdir())==[];checks.append(dict(label='no-rollback-after-failure'))
     for name in ('.tricu-stage-protected','x.tricu-stage-protected'):
         bad=b/name
         expected=tree_list([result(False,tree_list([text('invalid-destination'),text('stage/publish'),text('destination must name a file outside the private staging namespace'),text('not-published')])),result(False,tree_list([text('invalid-state'),text('stage/publish'),text('stage is not in the required phase'),text('none')]))])
         invoke('invalidDestination',[a,bad],name,expected)
         assert not bad.exists() and list(a.iterdir())==[]
     # Do not manufacture EXDEV through a mock here: require a distinct device.
     with tempfile.TemporaryDirectory(prefix='tricu-zig-exdev-',dir='/dev/shm') as foreign:
         foreign=Path(foreign)
         assert foreign.stat().st_dev != a.stat().st_dev,'EXDEV control needs distinct filesystems'
         output=foreign/'out'
         expected=tree_list([tree_list([text('io-error'),text('unknown')]),tree_list([text('invalid-state'),text('none')])])
         invoke('crossDevice',[a,output],'actual-exdev',expected)
         assert not output.exists() and list(a.iterdir())==[]
         flags=['--io']
         for i,v in enumerate((a,output)):
             arg=d/f'exdev-{i}.tree';arg.write_bytes(text(v));flags+=['--arg-tree-file',arg]
         assert r.invoke('exdev-haskell',pin,'crossDevice',flags)==expected
         assert not output.exists() and list(a.iterdir())==[]
     (d/'report.json').write_text(json.dumps(dict(checks=checks,haskell=r.records,probe_sha256=hashlib.sha256(args.probe.read_bytes()).hexdigest()),indent=2)+'\n')
     print(f'PASS {len(checks)} private staged-publication controls including real EXDEV')
 if __name__=='__main__':main()