io_v2_publication.py (7963B)
1 #!/usr/bin/env python3 2 """Source-owned exact length/digest policy over scoped staged writers and hashes.""" 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, field 10 from stable_bundle import byte_tree 11 from http_headers import number, result, tree_list 12 13 ROOT = Path(__file__).resolve().parent.parent 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 args.directory.mkdir(parents=True, exist_ok=False) 21 directory = args.directory.resolve() 22 runner = Runner(args.tricu, directory) 23 bundle = directory/'publication.arboricx' 24 runner.run('compile', ['compile', '--file', ROOT/'demos/io-v2/publish.tri', 25 '--store', runner.store, '--output', bundle]) 26 pin = runner.imported(bundle, 'import') 27 checks = [] 28 def invoke(label, name, values): 29 argv = ['--io'] 30 for index, value in enumerate(values): 31 arg = directory/f'{label}-{index}.tree' 32 arg.write_bytes(number(value) if isinstance(value, int) else tree_list([number(ord(c)) for c in str(value)])) 33 argv += ['--arg-tree-file', arg] 34 return runner.invoke(label, pin, name, argv) 35 callback = directory/'callback'; callback.mkdir() 36 assert invoke('rejected-destination', 'rejectedDestination', [callback]) == result(False, byte_tree(b'Destination rejected')) 37 failure = tree_list([byte_tree(b'invalid-response'), byte_tree(b'artifact-stream'), 38 byte_tree(b'destination'), byte_tree(b'Expected destination Result'), b'\0']) 39 assert invoke('malformed-destination', 'malformedDestination', [callback]) == tree_list([byte_tree(b'failed'), failure, b'\0']) 40 assert list(callback.iterdir()) == [] 41 checks += ['rejected-destination', 'malformed-destination'] 42 for size in (0, 1, 4095, 4096, 4097, 65536, 1048576): 43 body = (bytes(range(256))*((size+255)//256))[:size] 44 source = directory/f'input-{size}' 45 source.write_bytes(body) 46 target = directory/f'objects-{size}' 47 target.mkdir() 48 digest = hashlib.sha256(body).hexdigest() 49 for state in ('new', 'existing'): 50 label = f'{state}-{size}' 51 actual = invoke(label, 'publish', [source, target, size]) 52 assert actual == result(True, byte_tree(digest.encode())), label 53 assert (target/digest).read_bytes() == body, label 54 assert (target/digest).stat().st_mode & 0o077 == 0, label 55 assert sorted(p.name for p in target.iterdir()) == [digest], label 56 checks.append(label) 57 source, target = directory/'source-λ-树', directory/'objects-λ-树' 58 source.write_bytes(b'\0\xffbinary') 59 target.mkdir() 60 digest = hashlib.sha256(source.read_bytes()).hexdigest() 61 assert invoke('unicode-paths', 'publish', [source, target, 8]) == result(True, byte_tree(digest.encode())) 62 assert (target/digest).read_bytes() == source.read_bytes() 63 assert sorted(p.name for p in target.iterdir()) == [digest] 64 checks.append('unicode-paths') 65 source = directory/'negative-input' 66 source.write_bytes(b'abc') 67 digest = hashlib.sha256(b'abc').hexdigest() 68 for label, count, winner, message in ( 69 ('short', 4, None, 'Incomplete artifact stream'), 70 ('excess', 2, None, 'Artifact exceeds declared length'), 71 ('bad-winner', 3, b'xyz', 'Existing artifact digest mismatch'), 72 ('short-winner', 3, b'ab', 'Existing artifact length mismatch'), 73 ('long-winner', 3, b'abcd', 'Existing artifact length mismatch')): 74 target = directory/label 75 target.mkdir() 76 if winner is not None: 77 (target/digest).write_bytes(winner) 78 actual = invoke(label, 'publish', [source, target, count]) 79 assert actual == result(False, byte_tree(message.encode())), label 80 assert sorted(p.name for p in target.iterdir()) == ([] if winner is None else [digest]), label 81 if winner is not None: 82 assert (target/digest).read_bytes() == winner, label 83 checks.append(label) 84 for name, message in ( 85 ('earlySuccess', 'Incomplete artifact stream'), 86 ('ignoresWriteFailure', 'Artifact exceeds declared length'), 87 ('emptyChunk', 'Empty artifact chunk'), 88 ('overrideWrite', 'Incomplete artifact stream'), 89 ('failedAfterFull', 'Producer failed')): 90 target = directory/name 91 target.mkdir() 92 actual = invoke(name, name, [target]) 93 assert actual == result(False, byte_tree(message.encode())), name 94 assert list(target.iterdir()) == [], name 95 checks.append(name) 96 target = directory/'invalidChunk' 97 target.mkdir() 98 actual = invoke('invalidChunk', 'invalidChunk', [target]) 99 assert field(actual, 0) == b'\0' 100 assert field(actual, 1, 0, 0) == byte_tree(b'write-all') 101 assert field(actual, 1, 0, 1, 0) == number(0) 102 assert field(actual, 1, 0, 1, 1, 0, 0) == byte_tree(b'invalid-argument') 103 assert list(target.iterdir()) == [] 104 checks.append('invalidChunk') 105 for name, expected in [('cancelled', 'IO v2 task cancelled')]: 106 target = directory/name 107 target.mkdir() 108 try: 109 invoke(name, name, [target]) 110 except subprocess.CalledProcessError as error: 111 assert expected in error.output, error.output 112 else: 113 raise AssertionError(name+' resumed normally') 114 assert list(target.iterdir()) == [] 115 assert not (directory/'escape').exists() 116 checks.append(name) 117 target = directory/'badDigest' 118 target.mkdir() 119 expected = tree_list([byte_tree(b'failed'), tree_list([ 120 byte_tree(b'invalid-response'), byte_tree(b'artifact-stream'), byte_tree(b'digest'), 121 byte_tree(b'Invalid digest reply'), b'\0']), b'\0']) 122 assert invoke('badDigest', 'badDigest', [target]) == expected 123 assert list(target.iterdir()) == [] 124 assert not (directory/'escape').exists() 125 checks.append('badDigest') 126 for name in ('nestedProducer', 'forkedProducer'): 127 target = directory/name 128 target.mkdir() 129 expected = tree_list([byte_tree(b'failed'), tree_list([ 130 byte_tree(b'producer-boundary'), byte_tree(b'artifact-stream'), byte_tree(b'publish'), 131 byte_tree(b'Producer must be sequential'), 132 tree_list([byte_tree(b'outer'), byte_tree(b'inner')])]), b'\0']) 133 assert invoke(name, name, [target]) == expected, name 134 assert list(target.iterdir()) == [], name 135 checks.append(name) 136 observations = [dict(row) for row in runner.records if row['label'].endswith(('-65536', '-1048576'))] 137 for row in observations: 138 log = (directory/(row['label']+'.log')).read_text() 139 matched = re.search(r'([0-9,]+) bytes maximum residency', log) 140 assert matched, row 141 row['maximum_live_bytes'] = int(matched[1].replace(',', '')) 142 assert row['maximum_live_bytes'] <= 16*1024*1024, row 143 assert row['peak_kib'] <= 128*1024, row 144 paths = [Path(__file__), ROOT/'src/IOV2/Resources.hs', ROOT/'src/IOV2/Staging.hs', 145 ROOT/'src/IOV2/Digest.hs', ROOT/'src/IOV2/Driver.hs', ROOT/'src/IOV2/Protocol.hs', 146 ROOT/'test/source_order.py', ROOT/'test/rts_metrics.py', ROOT/'lib/io-v2.tri', 147 ROOT/'lib/io-v2/artifacts.tri', ROOT/'lib/handlers.tri', ROOT/'demos/io-v2/publish.tri'] 148 report = {'checks': checks, 'observations': observations, 'sources': { 149 str(path.resolve().relative_to(ROOT)): hashlib.sha256(path.read_bytes()).hexdigest() for path in paths}} 150 (directory/'report.json').write_text(json.dumps(report, indent=2)+'\n') 151 print(f'PASS: {len(checks)} publication and independent policy checks') 152 153 if __name__ == '__main__': 154 main()