tricu

io_v2_zig_publication.py (5486B)

---

 #!/usr/bin/env python3
 """Real source-owned publication policy on private Zig, with Haskell comparisons."""
 import argparse
 import hashlib
 import json
 import subprocess
 from pathlib import Path
 from source_order import Runner, field
 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/'publication.arboricx'
     r.run('compile',['compile','--file',ROOT/'demos/io-v2/publish.tri','--store',r.store,'--output',bundle]);pin=r.imported(bundle,'import')
     checks=[]
     def invoke(label,name,values,expected,compare=False):
         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=180)
         (d/(label+'.log')).write_text(output)
         actual=bytes(map(int,output.strip()))
         if expected is not None: assert actual==expected,(label,actual,expected)
         if compare:
             flags=['--io']
             for i,v in enumerate(values):
                 arg=d/f'{label}-{i}.tree';arg.write_bytes(number(v) if isinstance(v,int) else text(v));flags+=['--arg-tree-file',arg]
             assert r.invoke(label+'-haskell',pin,name,flags)==actual,label
         checks.append(dict(label=label,peak_kib=peak))
         return actual
     for size in (0,1,4095,4096,4097,65536,1048576):
         source=d/f'input-{size}';body=(bytes(range(256))*((size+255)//256))[:size];source.write_bytes(body)
         target=d/f'objects-{size}';target.mkdir();digest=hashlib.sha256(body).hexdigest()
         for state in ('new','existing'):
             invoke(f'{state}-{size}','publish',[source,target,size],result(True,text(digest)),size==4097)
             assert (target/digest).read_bytes()==body
             assert (target/digest).stat().st_mode&0o077==0
             assert sorted(p.name for p in target.iterdir())==[digest]
     source=d/'source-λ-树';source.write_bytes(b'\0\xffbinary');target=d/'objects-λ-树';target.mkdir()
     digest=hashlib.sha256(source.read_bytes()).hexdigest()
     invoke('unicode','publish',[source,target,8],result(True,text(digest)),True)
     assert (target/digest).read_bytes()==source.read_bytes()
     source=d/'negative-input';source.write_bytes(b'abc');digest=hashlib.sha256(b'abc').hexdigest()
     for label,count,winner,message in [('short',4,None,'Incomplete artifact stream'),('excess',2,None,'Artifact exceeds declared length'),('bad-winner',3,b'xyz','Existing artifact digest mismatch'),('short-winner',3,b'ab','Existing artifact length mismatch'),('long-winner',3,b'abcd','Existing artifact length mismatch')]:
         target=d/label;target.mkdir()
         if winner is not None:(target/digest).write_bytes(winner)
         invoke(label,'publish',[source,target,count],result(False,text(message)),True)
         assert sorted(p.name for p in target.iterdir())==([] if winner is None else [digest])
         if winner is not None:assert (target/digest).read_bytes()==winner
     for name,message in [('earlySuccess','Incomplete artifact stream'),('ignoresWriteFailure','Artifact exceeds declared length'),('emptyChunk','Empty artifact chunk'),('overrideWrite','Incomplete artifact stream'),('failedAfterFull','Producer failed'),('rejectedDestination','Destination rejected')]:
         target=d/name;target.mkdir();invoke(name,name,[target],result(False,text(message)),True);assert list(target.iterdir())==[]
     for name,problem in [
         ('malformedDestination',tree_list([text('invalid-response'),text('artifact-stream'),text('destination'),text('Expected destination Result'),b'\0'])),
         ('badDigest',tree_list([text('invalid-response'),text('artifact-stream'),text('digest'),text('Invalid digest reply'),b'\0'])),
         *[(n,tree_list([text('producer-boundary'),text('artifact-stream'),text('publish'),text('Producer must be sequential'),tree_list([text('outer'),text('inner')])])) for n in ('nestedProducer','forkedProducer')]]:
         target=d/name;target.mkdir();invoke(name,name,[target],tree_list([text('failed'),problem,b'\0']),True);assert list(target.iterdir())==[]
     target=d/'invalidChunk';target.mkdir();actual=invoke('invalidChunk','invalidChunk',[target],None,True)
     assert field(actual,0)==b'\0' and field(actual,1,0,0)==text('write-all') and field(actual,1,0,1,0)==number(0)
     assert field(actual,1,0,1,1,0,0)==text('invalid-argument');assert list(target.iterdir())==[]
     target=d/'cancelled';target.mkdir()
     process=subprocess.run([str(args.probe.resolve()),str(bundle),'cancelled','text:'+str(target)],capture_output=True,timeout=60)
     assert process.returncode==1 and b'cancelled (reason not forced)' in process.stderr
     assert list(target.iterdir())==[] and not (d/'escape').exists()
     checks.append(dict(label='cancelled-no-private-names'))
     (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 Zig source publication/policy controls')
 if __name__=='__main__':main()