import argparse import hashlib from pathlib import Path VALUES = { 'conservative': { 'remote_assignment_reserve_bytes': 64 * 1024 * 1024, 'remote_assignment_max_active': 1, 'keycheck_queue_max_items': 8192, 'keycheck_queue_max_bytes': 64 * 1024 * 1024, }, 'final': { 'remote_assignment_reserve_bytes': 2 * 1024 * 1024, 'remote_assignment_max_active': 50, 'keycheck_queue_max_items': 131072, 'keycheck_queue_max_bytes': 128 * 1024 * 1024, }, } def render(payload, mode): text = payload.decode('utf-8', errors='strict') if '\r' in text: raise ValueError('configuration must use LF line endings') lines = text.splitlines(keepends=True) values = VALUES[mode] hard_limit = ' result_bundle_max_event_bytes: 67108864\n' if lines.count(hard_limit) != 1: raise ValueError('unexpected hard result limit') for field in ('remote_assignment_reserve_bytes', 'remote_assignment_max_active'): if any(line.startswith(f' {field}:') for line in lines): raise ValueError(f'active configuration already contains {field}') index = lines.index(hard_limit) + 1 lines[index:index] = [ f" remote_assignment_reserve_bytes: {values['remote_assignment_reserve_bytes']}\n", f" remote_assignment_max_active: {values['remote_assignment_max_active']}\n", ] replacements = { ' keycheck_queue_max_items: 8192\n': ( f" keycheck_queue_max_items: {values['keycheck_queue_max_items']}\n" ), ' keycheck_queue_max_bytes: 67108864\n': ( f" keycheck_queue_max_bytes: {values['keycheck_queue_max_bytes']}\n" ), } for before, after in replacements.items(): if lines.count(before) != 1: raise ValueError(f'unexpected active configuration field: {before.strip()}') lines[lines.index(before)] = after return ''.join(lines).encode('utf-8') def main(): parser = argparse.ArgumentParser() parser.add_argument('--input', required=True) parser.add_argument('--output', required=True) parser.add_argument('--mode', choices=tuple(VALUES), required=True) parser.add_argument('--expected-sha256', required=True) args = parser.parse_args() source = Path(args.input) destination = Path(args.output) payload = source.read_bytes() if hashlib.sha256(payload).hexdigest() != args.expected_sha256: raise SystemExit('active configuration identity changed') rendered = render(payload, args.mode) destination.write_bytes(rendered) print(hashlib.sha256(rendered).hexdigest()) if __name__ == '__main__': main()