io_positioned.py (5601B)
1 #!/usr/bin/env python3 2 """Independent complete byte/result oracles for positioned file compositions.""" 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, pair, result, tree_list 9 from stable_bundle import byte_tree 10 ROOT = Path(__file__).resolve().parent.parent 11 12 def text(value): 13 return tree_list([number(ord(c)) for c in value]) 14 15 def main(): 16 parser = argparse.ArgumentParser(description=__doc__) 17 parser.add_argument('tricu', type=Path) 18 parser.add_argument('--directory', type=Path, required=True) 19 args = parser.parse_args() 20 directory = args.directory.resolve() 21 directory.mkdir(parents=True, exist_ok=False) 22 runner = Runner(args.tricu, directory) 23 source = ROOT/'demos/io/positioned.tri' 24 bundle = directory/'positioned.arboricx' 25 runner.run('compile', ['compile', '--file', source, '--store', runner.store, '--output', bundle]) 26 pin = runner.imported(bundle, 'import') 27 checks = [] 28 29 def invoke(label, name, values): 30 argv = ['--io'] 31 for index, value in enumerate(values): 32 path = directory/f'{label}-{index}.tree' 33 path.write_bytes(value) 34 argv += ['--arg-tree-file', path] 35 return runner.invoke(label, pin, name, argv) 36 37 for index, (original, offset, patch, length, exists) in enumerate([ 38 (b'abcdefgh', 2, bytes([0,255,13,10]), 8, True), 39 (b'', 5, b'X', 12, False), 40 (b'0123456789', 3, b'ABCDE', 4, True), 41 (b'unchanged prefix', 0, b'', 16, True), 42 (b'discarded', 2, b'xy', 0, True), 43 (bytes(range(256))*4096, 524288, bytes(range(128)), 1048585, True), 44 ]): 45 path = directory/f'patch-{index}-λ.bin' 46 if exists: 47 path.write_bytes(original) 48 wanted = bytearray(original) 49 if patch: 50 end = offset+len(patch) 51 if end > len(wanted): 52 wanted.extend(bytes(end-len(wanted))) 53 wanted[offset:end] = patch 54 if length > len(wanted): 55 wanted.extend(bytes(length-len(wanted))) 56 del wanted[length:] 57 prefix = pair(text('data'), byte_tree(wanted[:4096])) if wanted else pair(text('eof'), b'\0') 58 expected = tree_list([number(n) for n in 59 [len(original), offset, offset+len(patch), offset+len(patch), length]]+[prefix]) 60 actual = invoke(f'patch-{index}', 'patch', [text(str(path)), number(offset), byte_tree(patch), number(length)]) 61 assert actual == expected, index 62 assert path.read_bytes() == wanted, index 63 checks.append(dict(label=f'patch-{index}', sha256=hashlib.sha256(wanted).hexdigest())) 64 for index, original in enumerate([None, b'', b'prior\x00\xff', bytes(range(256))*4096]): 65 path = directory/f'append-{index}.bin' 66 if original is not None: 67 path.write_bytes(original) 68 body = bytes(range(256))*16 69 before = original or b'' 70 actual = invoke(f'append-{index}', 'append', [text(str(path)), byte_tree(body)]) 71 assert actual == tree_list([number(len(before)),number(len(before)+len(body))]), index 72 assert path.read_bytes() == before+body, index 73 checks.append(dict(label=f'append-{index}', sha256=hashlib.sha256(before+body).hexdigest())) 74 # Exercise the wire's wide offset without materializing a multi-GiB value: 75 # write one sparse byte, shrink, and compare the entire resulting file. 76 path = directory/'wide.bin' 77 actual = invoke('wide-offset', 'patch', [text(str(path)),number(4294967296),byte_tree(b'X'),number(1)]) 78 assert actual == tree_list([number(n) for n in [0,4294967296,4294967297,4294967297,1]]+ 79 [pair(text('data'),byte_tree(b'\0'))]) 80 assert path.read_bytes() == b'\0' 81 checks.append(dict(label='wide-offset')) 82 path = directory/'guard.bin' 83 path.write_bytes(b'keep me') 84 def problem(category, operation, detail=''): 85 return result(False, tree_list([text(v) for v in [category, operation, detail, 'none']])) 86 expected = tree_list([problem('unsupported','file/'+op,'operation unavailable in this file mode') 87 for op in ['seek','position','resize']]+[problem('wrong-kind','stream/read')]) 88 assert invoke('append-guards', 'appendGuards', [text(str(path))]) == expected 89 assert path.read_bytes() == b'keep me' 90 checks.append(dict(label='append-guards')) 91 for index, value in enumerate([number(9223372036854775808), pair(b'\0',b'\0'), b'\1\0']): 92 assert invoke(f'offset-{index}', 'badOffset', [text(str(path)),value]) == problem('invalid-argument','file/seek') 93 assert path.read_bytes() == b'keep me' 94 checks.append(dict(label=f'offset-{index}')) 95 observations = [row for row in runner.records if row['label'].startswith(('patch-','append-'))] 96 for row in observations: 97 assert row['peak_kib'] <= 128*1024, row 98 assert row['rts_process']['maximum_live_bytes'] <= 16*1024*1024, row 99 paths = [source, Path(__file__).resolve(), ROOT/'lib/io.tri', ROOT/'src/IO/Resources.hs', 100 ROOT/'src/IO/Driver.hs', ROOT/'src/IO/Protocol.hs', ROOT/'test/source_order.py', 101 ROOT/'test/rts_metrics.py', ROOT/'bootstrap/build-stable.arboricx'] 102 (directory/'report.json').write_text(json.dumps(dict(checks=checks, observations=observations, 103 sources={str(p.relative_to(ROOT)):hashlib.sha256(p.read_bytes()).hexdigest() for p in paths}), indent=2)+'\n') 104 print(f'PASS: {len(checks)} independent positioned-file checks') 105 106 if __name__ == '__main__': 107 main()