io_text.py (7344B)
1 #!/usr/bin/env python3 2 """Independent UTF-8 boundaries, exact source provenance and text pipelines.""" 3 import argparse 4 import hashlib 5 import json 6 from pathlib import Path 7 from source_order import Runner 8 from http_headers import number, result, pair, tree_list 9 from stable_bundle import byte_tree 10 from process_rss import run_guarded 11 12 ROOT = Path(__file__).resolve().parent.parent 13 14 def text(s): 15 return tree_list([number(ord(c)) for c in s]) 16 17 def problem(category, op, detail, progress='none'): 18 return tree_list([text(category),text(op),text(detail),text(progress)]) 19 20 def main(): 21 parser = argparse.ArgumentParser(description=__doc__) 22 parser.add_argument('tricu', type=Path) 23 parser.add_argument('--directory', type=Path, required=True) 24 args = parser.parse_args() 25 directory = args.directory.resolve() 26 directory.mkdir(parents=True, exist_ok=False) 27 runner = Runner(args.tricu,directory) 28 source = ROOT/'demos/io/text.tri' 29 bundle = directory/'text.arboricx' 30 runner.run('compile',['compile','--file',source,'--store',runner.store,'--output',bundle]) 31 pin = runner.imported(bundle,'import') 32 checks = [] 33 def invoke(label,name,values,expected): 34 argv = ['--io'] 35 for i,value in enumerate(values): 36 path = directory/f'{label}-{i}.tree' 37 path.write_bytes(value) 38 argv += ['--arg-tree-file',path] 39 assert runner.invoke(label,pin,name,argv) == expected,label 40 checks.append(label) 41 for i,value in enumerate(['','ascii\r\n\0','位馃榾e\u0301茅','\ufeff','\uffff\U0010ffff','馃榾'*1025,'a'*1024+'馃榾'+'b'*1024]): 42 payload = value.encode('utf-8') 43 invoke(f'encode-{i}','encode',[number(len(payload)),text(value)],result(True,byte_tree(payload))) 44 output = directory/f'write-{i}.bin' 45 invoke(f'write-{i}','write',[text(str(output)),text(value)],result(True,b'\0')) 46 assert output.read_bytes() == payload 47 for i,payload in enumerate([b'',b'a\r\n\0', '位馃榾e\u0301茅'.encode(), 48 b'a'*4092+'馃榾'.encode(), b'a'*4091+'馃榾x'.encode(), 49 ('a\ufeffe\u0301\r\n'*100).encode()]): 50 path = directory/f'source-{i}.bin' 51 path.write_bytes(payload) 52 invoke(f'source-{i}','source',[text(str(path)),number(len(payload))], 53 result(True,pair(byte_tree(payload),text(payload.decode('utf-8'))))) 54 assert path.read_bytes() == payload 55 # Confirm accounting at the 4093-byte read boundary and its one-byte EOF 56 # probe, including multiple chunks. These bytes are independently decoded. 57 for size in [4092,4093,4094,8186,8187]: 58 payload=b'a'*size 59 path=directory/f'count-boundary-{size}.bin' 60 path.write_bytes(payload) 61 invoke(f'count-exact-{size}','source',[text(str(path)),number(size)], 62 result(True,pair(byte_tree(payload),text(payload.decode())))) 63 invoke(f'count-short-{size}','source',[text(str(path)),number(size-1)], 64 result(False,tree_list([text('read-source'),number(size), 65 problem('invalid-bound','text/read-source','source exceeds byte limit')]))) 66 payload = 'A馃榾位'.encode() 67 # Every split through the multibyte suffix, with independent expected carry. 68 for i in range(len(payload)+1): 69 prefix = payload[:i] 70 import codecs 71 decoder = codecs.getincrementaldecoder('utf-8')('strict') 72 decoded = decoder.decode(prefix, final=False) 73 carry = decoder.getstate()[0] 74 invoke(f'prefix-{i}','decode',[number(0),byte_tree(prefix)], 75 result(True,tree_list([text(decoded),byte_tree(carry)]))) 76 for i,(payload,cause) in enumerate([ 77 (b'\xef\xbb\xbfabc',problem('invalid-data','text/read-source','initial source BOM is not permitted')), 78 (b'good\xff',problem('invalid-data','utf8/decode','invalid UTF-8 leading byte at byte 4')), 79 (b'\xe2\x82',problem('invalid-data','utf8/decode','incomplete UTF-8 sequence at byte 0')), 80 (b'\xed\xa0\x80',problem('invalid-data','utf8/decode','invalid UTF-8 sequence at byte 0'))]): 81 path = directory/f'invalid-{i}.bin'; path.write_bytes(payload) 82 invoke(f'invalid-source-{i}','source',[text(str(path)),number(len(payload))], 83 result(False,tree_list([text('read-source'),number(len(payload)),cause]))) 84 path = directory/'excess.bin'; path.write_bytes(b'x') 85 invoke('source-limit','source',[text(str(path)),number(0)],result(False,tree_list([ 86 text('read-source'),number(1),problem('invalid-bound','text/read-source','source exceeds byte limit')]))) 87 invoke('encode-limit','encode',[number(1),text('位')],result(False, 88 problem('invalid-bound','text/encode','encoded text exceeds byte limit'))) 89 invoke('surrogate','encode',[number(4096),tree_list([number(0xd800)])],result(False, 90 problem('invalid-argument','utf8/encode',''))) 91 partial = directory/'partial.bin' 92 invalid_text = tree_list([number(97)]*1024+[number(0xd800)]) 93 invoke('write-invalid-tail','write',[text(str(partial)),invalid_text],result(False,tree_list([ 94 text('write-text'),number(1024),problem('invalid-argument','utf8/encode','')]))) 95 assert partial.read_bytes() == b'a'*1024 96 for size in [4096,65536]: 97 payload = (b'ab\r\n\0'+ '位馃榾'.encode())*(size//11) 98 input_path = directory/f'pipeline-{size}.bin'; output = directory/f'pipeline-{size}-out.bin' 99 input_path.write_bytes(payload) 100 invoke(f'pipeline-{size}','roundTrip',[text(str(input_path)),text(str(output)),number(len(payload))],result(True,b'\0')) 101 assert output.read_bytes() == payload 102 # Independent raw-protocol boundaries; excess heads AND tails diverge. 103 # Reject without demanding excess cells or writing a prefix. 104 boundary = directory/'boundaries.arboricx' 105 runner.run('compile-boundaries',['compile','--file',ROOT/'test/io/utf8-boundaries.tri','--store',runner.store,'--output',boundary]) 106 original_bundle, original_pin = bundle, pin 107 bundle, pin = boundary, runner.imported(boundary,'import-boundaries') 108 for name,operation in [('excessEncode','utf8/encode'),('excessDecode','utf8/decode'),('excessDigest','digest/update')]: 109 invoke(name,name,[number(0)],result(False,problem('invalid-argument',operation,''))) 110 path = directory/'excess-write' 111 invoke('excessWrite','excessWrite',[text(str(path)),number(0)],result(False,problem('invalid-argument','stream/write',''))) 112 assert path.read_bytes() == b'' 113 bundle, pin = original_bundle, original_pin 114 observations = [r for r in runner.records if r['label'].startswith('pipeline-')] 115 for row in observations: 116 assert row['peak_kib'] <= 256*1024,row 117 assert row['rts_process']['maximum_live_bytes'] <= 128*1024*1024,row 118 files = ['lib/io.tri','lib/io/text.tri','demos/io/text.tri','test/io_text.py', 119 'test/source_order.py','test/rts_metrics.py','src/IO/UTF8.hs','src/IO/Driver.hs', 120 'src/IO/Protocol.hs','bootstrap/build-stable.arboricx'] 121 (directory/'report.json').write_text(json.dumps(dict(checks=checks,observations=observations, 122 123 source_hashes={p:hashlib.sha256((ROOT/p).read_bytes()).hexdigest() for p in files}),indent=2)+'\n') 124 print(f'{len(checks)} independent text/UTF-8 checks passed'+('')) 125 126 if __name__ == '__main__': main()