scripts/prepare-controlled-upgrade.test.py

Rama main · réplica de lectura del repositorio original.

import base64
import copy
import gzip
import hashlib
import json
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile
import unittest

PREPARE = Path(__file__).with_name('prepare-controlled-upgrade.py')
PATHS = [
    'scripts/verify-node-controlled-upgrade.mjs',
    'scripts/verify-node-controlled-upgrade.test.mjs',
    'scripts/verify-node-direct-registration.mjs',
    'scripts/fixtures/node-controlled-upgrade-core_test.go',
    'scripts/fixtures/node-controlled-upgrade-provider.mjs',
    'scripts/fixtures/node-upgrade-old-binaries.json',
]

def fixture():
    files = []
    for path in PATHS:
        content = ('synthetic ' + path).encode()
        if path.endswith('node-upgrade-old-binaries.json'):
            content = json.dumps({'core_commit': 'c'*40}).encode()
        blob = b'blob ' + str(len(content)).encode() + b'\0' + content
        files.append(dict(path=path, mode='100755' if 'provider.mjs' in path else '100644',
                          git_blob=hashlib.sha1(blob).hexdigest(), sha256=hashlib.sha256(content).hexdigest(),
                          content=base64.b64encode(content).decode()))
    return dict(root_commit='a'*40, files=files)

class Preparation(unittest.TestCase):
    def run_case(self, mutate=lambda bundle, lock: None):
        with tempfile.TemporaryDirectory(prefix='node-fixture-test-') as directory:
            root = Path(directory)
            (root / 'scripts').mkdir()
            (root / 'tools/controlled-upgrade').mkdir(parents=True)
            shutil.copyfile(PREPARE, root / 'scripts/prepare-controlled-upgrade.py')
            bundle = fixture()
            lock = dict(root_commit=bundle['root_commit'], core_commit='c'*40,
                        files=[{k:v for k,v in f.items() if k != 'content'} for f in bundle['files']])
            mutate(bundle, lock)
            archive = gzip.compress(json.dumps(bundle).encode())
            lock.setdefault('sha256', hashlib.sha256(archive).hexdigest())
            (root / 'tools/controlled-upgrade/root-fixture.json.gz').write_bytes(archive)
            (root / 'tools/controlled-upgrade/fixture-lock.json').write_text(json.dumps(lock))
            target = root / 'output'
            result = subprocess.run([sys.executable, str(root / 'scripts/prepare-controlled-upgrade.py'), str(target)], capture_output=True)
            if result.returncode == 0:
                self.assertEqual({p.relative_to(target).as_posix() for p in target.rglob('*') if p.is_file()}, set(PATHS))
                self.assertEqual((target / PATHS[4]).stat().st_mode & 0o777, 0o755)
            else:
                self.assertFalse(target.exists(), 'all validation must precede writes')
            return result.returncode

    def test_valid_export_and_executable_provider(self):
        self.assertEqual(self.run_case(), 0)

    def test_corrupt_hash_and_unreviewed_content_rejected(self):
        mutations = [
            lambda b,l: l.update(sha256='0'*64),
            lambda b,l: b.update(root_commit='b'*40),
            lambda b,l: b['files'][0].update(content=base64.b64encode(b'changed').decode()),
            lambda b,l: b['files'][0].update(mode='120000'),
            lambda b,l: b['files'].pop(),
            lambda b,l: b['files'].append(copy.deepcopy(b['files'][0])),
            lambda b,l: b['files'][0].update(path='../escaped'),
            lambda b,l: b['files'][0].update(path='/tmp/escaped'),
            lambda b,l: l['files'][0].update(git_blob='0'*40),
            lambda b,l: l.update(core_commit='d'*40),
            lambda b,l: l.pop('core_commit'),
            lambda b,l: l.update(core_commit='main'),
        ]
        for mutate in mutations:
            with self.subTest(mutate=mutate):
                self.assertNotEqual(self.run_case(mutate), 0)

    def test_matching_metadata_cannot_authorize_invalid_mode_or_content(self):
        for field, value in [('mode', '120000'), ('sha256', '0'*64), ('git_blob', '0'*40)]:
            def mutate(bundle, lock):
                bundle['files'][0][field] = value
                lock['files'][0][field] = value
            with self.subTest(field=field):
                self.assertNotEqual(self.run_case(mutate), 0)

if __name__ == '__main__':
    unittest.main()