Function copy_selected

scripts/ci/select-release-candidates.py:446–488

(
    source: pathlib.Path,
    destination: pathlib.Path,
    *,
    expected_sha256: str,
    expected_size: int,
)

def copy_selected(
    source: pathlib.Path,
    destination: pathlib.Path,
    *,
    expected_sha256: str,
    expected_size: int,
) -> None:
    if sha256_file(source, expected_size=expected_size) != expected_sha256:
        raise ContractError(f"candidate object changed before selection copy: {source}")
    destination.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
    before = regular_status(source, ceiling=MAX_CANDIDATE_BYTES, label="candidate object")
    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
    descriptor = os.open(source, flags)
    try:
        opened = os.fstat(descriptor)
        if not stat.S_ISREG(opened.st_mode) or (opened.st_dev, opened.st_ino) != (
            before.st_dev,
            before.st_ino,
        ):
            raise ContractError(f"candidate object changed while being opened: {source}")
        digest = hashlib.sha256()
        total = 0
        with os.fdopen(descriptor, "rb", closefd=False) as input_handle, destination.open(
            "xb"
        ) as output_handle:
            while True:
                chunk = input_handle.read(1024 * 1024)
                if not chunk:
                    break
                total += len(chunk)
                if total > expected_size:
                    raise ContractError(f"candidate object grew during selection: {source}")
                digest.update(chunk)
                output_handle.write(chunk)
            output_handle.flush()
            os.fsync(output_handle.fileno())
    finally:
        os.close(descriptor)
    if total != expected_size or digest.hexdigest() != expected_sha256:
        raise ContractError(f"candidate object changed during selection copy: {source}")
    destination.chmod(0o555)
    if sha256_file(destination, expected_size=expected_size) != expected_sha256:
        raise ContractError(f"selected copy is not content-bound: {destination}")


def main(argv: Sequence[str]) -> None: