io_v2_objects.py (6705B)
1 #!/usr/bin/env python3 2 """Independent source-owned CAS bytes, exact winners and shared-tree transfer.""" 3 import argparse 4 import hashlib 5 import json 6 from pathlib import Path 7 import re 8 import subprocess 9 from source_order import Runner 10 from native_indexed_store import indexed_bytes, digest 11 from executable_object import INDEXED 12 from io_v2_indexed_write import dag 13 from http_headers import number, result, tree_list 14 15 ROOT = Path(__file__).resolve().parent.parent 16 17 def text(s): 18 return tree_list([number(ord(c)) for c in s]) 19 20 def problem(category, op, detail, progress): 21 return result(False, tree_list([text(category),text(op),text(detail),text(progress)])) 22 23 def main(): 24 parser = argparse.ArgumentParser(description=__doc__) 25 parser.add_argument('tricu', type=Path) 26 parser.add_argument('--directory', type=Path, required=True) 27 args = parser.parse_args() 28 directory = args.directory.resolve() 29 directory.mkdir(parents=True, exist_ok=False) 30 runner = Runner(args.tricu, directory) 31 source = ROOT/'demos/io-v2/objects.tri' 32 bundle = directory/'objects.arboricx' 33 runner.run('compile',['compile','--file',source,'--store',runner.store,'--output',bundle]) 34 module = runner.imported(bundle,'import') 35 checks = [] 36 def options(label, values): 37 argv = ['--io'] 38 for i,value in enumerate(values): 39 path = directory/f'{label}-{i}.tree' 40 path.write_bytes(value) 41 argv += ['--arg-tree-file',path] 42 return argv 43 def invoke(label,name,values,expected): 44 assert runner.invoke(label,module,name,options(label,values)) == expected,label 45 checks.append(label) 46 def clean(root): 47 assert not list(root.rglob('.tricu-*')),root 48 for i,tree in enumerate([b'\0',b'\1\0',number(42)]): 49 root = directory/f'root-{i}' 50 root.mkdir() 51 payload = indexed_bytes(tree) 52 pin = digest(INDEXED,payload) 53 path = root/'objects'/pin[:3]/pin 54 for label in ['new','existing']: 55 invoke(f'{label}-{i}','put',[text(str(root)),number(len(payload)),tree],result(True,text(pin))) 56 assert path.read_bytes() == payload 57 assert sorted(p.relative_to(root).as_posix() for p in root.rglob('*') if p.is_file()) == ['objects/'+pin[:3]+'/'+pin] 58 clean(root) 59 invoke(f'get-{i}','get',[text(str(root)),number(len(payload)),text(pin)],result(True,tree)) 60 path.write_bytes(b'X'+payload[1:]) 61 invoke(f'corrupt-put-{i}','put',[text(str(root)),number(len(payload)),tree], 62 problem('integrity-failed','indexed-tree/verify','indexed frame differs from canonical tree','unknown')) 63 invoke(f'corrupt-get-{i}','get',[text(str(root)),number(len(payload)),text(pin)], 64 problem('integrity-failed','indexed-tree/read','CAS digest mismatch','unknown')) 65 assert path.read_bytes() == b'X'+payload[1:] 66 clean(root) 67 path.write_bytes(payload+b'extra') 68 invoke(f'excess-{i}','put',[text(str(root)),number(len(payload)),tree], 69 problem('invalid-data','objects/verify','trailing bytes','unknown')) 70 assert path.read_bytes() == payload+b'extra' 71 clean(root) 72 absent = directory/'must-not-be-created' 73 for i,invalid in enumerate(['../outside','A'*64,'0'*63,'0'*65,'λ'*64,'\0'*64]): 74 invoke(f'invalid-{i}','get',[text(str(absent)),number(65536),text(invalid)], 75 problem('invalid-argument','objects/get','invalid CAS digest','none')) 76 for i, invalid in enumerate([b'\1\0', tree_list([number(48)[:-1]+b'\2\0\0']*64)]): 77 invoke(f'malformed-digest-{i}','get',[text(str(absent)),number(65536),invalid], 78 problem('invalid-argument','objects/get','invalid CAS digest','none')) 79 assert not absent.exists() 80 invoke('absent-root','putCategory',[text(str(absent)),number(65536),b'\0'],text('not-found')) 81 blocked = directory/'blocked'; blocked.mkdir(); (blocked/'objects').write_bytes(b'keep') 82 invoke('blocked','putCategory',[text(str(blocked)),number(65536),b'\0'],text('not-directory')) 83 assert (blocked/'objects').read_bytes() == b'keep' 84 limited = directory/'limited'; limited.mkdir() 85 invoke('limit','putCategory',[text(str(limited)),number(0),b'\0'],text('invalid-bound')) 86 assert list(limited.iterdir()) == [] 87 source_root = directory/'source'; target = directory/'target' 88 source_root.mkdir(); target.mkdir() 89 payload = dag(200,True) 90 pin = digest(INDEXED,payload) 91 path = source_root/'objects'/pin[:3]/pin 92 path.parent.mkdir(parents=True); path.write_bytes(payload) 93 invoke('shared-copy','copy',[text(str(source_root)),text(str(target)),number(len(payload)),text(pin)],result(True,text(pin))) 94 assert (target/'objects'/pin[:3]/pin).read_bytes() == payload 95 clean(source_root); clean(target) 96 # Two independent processes publish to one new namespace. Keep complete 97 # output/byte checks; this is a correctness race, not a capacity benchmark. 98 race = directory/'race'; race.mkdir() 99 tree = number(19); payload = indexed_bytes(tree); pin = digest(INDEXED,payload) 100 argv = options('race',[text(str(race)),number(len(payload)),tree]) 101 command = [str(args.tricu.resolve()),'arboricx','run','--module',module,'--name','put','--store',str(runner.store),'--format','ternary',*map(str,argv),'+RTS','-N1','-M256m','-RTS'] 102 workers = [subprocess.Popen(command,stdout=subprocess.PIPE,stderr=subprocess.PIPE) for _ in range(2)] 103 try: 104 for worker in workers: 105 out,err = worker.communicate(timeout=120) 106 assert worker.returncode == 0,(out,err) 107 raw = re.search(rb'(?m)^([012]+)$',out)[1] 108 assert bytes(int(c)-48 for c in raw) == result(True,text(pin)),out 109 finally: 110 for worker in workers: 111 if worker.poll() is None: worker.kill() 112 worker.wait() 113 assert (race/'objects'/pin[:3]/pin).read_bytes() == payload 114 clean(race); checks.append('concurrent-publish') 115 files = ['lib/io-v2/objects.tri','lib/io-v2.tri','demos/io-v2/objects.tri','test/io_v2_objects.py', 116 'test/native_indexed_store.py','test/io_v2_indexed_write.py','test/executable_object.py', 117 'test/source_order.py','test/rts_metrics.py','src/IOV2/Resources.hs','src/IOV2/Descriptor.hs', 118 'src/IOV2/Filesystem.hs','src/IOV2/Codec.hs','bootstrap/build-stable.arboricx'] 119 (directory/'report.json').write_text(json.dumps(dict(checks=checks,observations=runner.records, 120 source_hashes={p:hashlib.sha256((ROOT/p).read_bytes()).hexdigest() for p in files}),indent=2)+'\n') 121 print(f'{len(checks)} source CAS controls passed') 122 123 if __name__ == '__main__': main()