mirror of
https://github.com/saltstack/salt.git
synced 2025-04-16 09:40:20 +00:00
119 lines
3.5 KiB
Python
119 lines
3.5 KiB
Python
# pylint: disable=resource-leakage,broad-except
|
|
from __future__ import annotations
|
|
|
|
import pathlib
|
|
|
|
import packaging.version
|
|
from ptscripts import Context
|
|
from rich.progress import (
|
|
BarColumn,
|
|
Column,
|
|
DownloadColumn,
|
|
Progress,
|
|
TextColumn,
|
|
TimeRemainingColumn,
|
|
TransferSpeedColumn,
|
|
)
|
|
|
|
REPO_ROOT = pathlib.Path(__file__).resolve().parent.parent
|
|
GPG_KEY_FILENAME = "SALT-PROJECT-GPG-PUBKEY-2023"
|
|
NIGHTLY_BUCKET_NAME = "salt-project-prod-salt-artifacts-nightly"
|
|
STAGING_BUCKET_NAME = "salt-project-prod-salt-artifacts-staging"
|
|
RELEASE_BUCKET_NAME = "salt-project-prod-salt-artifacts-release"
|
|
BACKUP_BUCKET_NAME = "salt-project-prod-salt-artifacts-backup"
|
|
|
|
|
|
class UpdateProgress:
|
|
def __init__(self, progress, task):
|
|
self.progress = progress
|
|
self.task = task
|
|
|
|
def __call__(self, chunk_size):
|
|
self.progress.update(self.task, advance=chunk_size)
|
|
|
|
|
|
def create_progress_bar(file_progress: bool = False, **kwargs):
|
|
if file_progress:
|
|
return Progress(
|
|
TextColumn("[progress.description]{task.description}"),
|
|
BarColumn(),
|
|
DownloadColumn(),
|
|
TransferSpeedColumn(),
|
|
TextColumn("eta"),
|
|
TimeRemainingColumn(),
|
|
**kwargs,
|
|
)
|
|
return Progress(
|
|
TextColumn(
|
|
"[progress.description]{task.description}", table_column=Column(ratio=3)
|
|
),
|
|
BarColumn(),
|
|
expand=True,
|
|
**kwargs,
|
|
)
|
|
|
|
|
|
def export_gpg_key(ctx: Context, key_id: str, export_path: pathlib.Path):
|
|
keyfile_gpg = export_path.joinpath(GPG_KEY_FILENAME).with_suffix(".gpg")
|
|
if keyfile_gpg.exists():
|
|
keyfile_gpg.unlink()
|
|
ctx.info(f"Exporting GnuPG Key '{key_id}' to {keyfile_gpg} ...")
|
|
ctx.run("gpg", "--output", str(keyfile_gpg), "--export", key_id)
|
|
keyfile_pub = export_path.joinpath(GPG_KEY_FILENAME).with_suffix(".pub")
|
|
if keyfile_pub.exists():
|
|
keyfile_pub.unlink()
|
|
ctx.info(f"Exporting GnuPG Key '{key_id}' to {keyfile_pub} ...")
|
|
ctx.run("gpg", "--armor", "--output", str(keyfile_pub), "--export", key_id)
|
|
|
|
|
|
def gpg_sign(ctx: Context, key_id: str, path: pathlib.Path):
|
|
ctx.info(f"GPG Signing '{path}' ...")
|
|
signature_fpath = path.parent / f"{path.name}.asc"
|
|
if signature_fpath.exists():
|
|
signature_fpath.unlink()
|
|
ctx.run(
|
|
"gpg",
|
|
"--local-user",
|
|
key_id,
|
|
"--output",
|
|
str(signature_fpath),
|
|
"--armor",
|
|
"--detach-sign",
|
|
"--sign",
|
|
str(path),
|
|
)
|
|
|
|
|
|
class Version(packaging.version.Version):
|
|
def __lt__(self, other):
|
|
if not isinstance(other, self.__class__):
|
|
other = self.__class__(other)
|
|
return super().__lt__(other)
|
|
|
|
def __le__(self, other):
|
|
if not isinstance(other, self.__class__):
|
|
other = self.__class__(other)
|
|
return super().__le__(other)
|
|
|
|
def __eq__(self, other):
|
|
if not isinstance(other, self.__class__):
|
|
other = self.__class__(other)
|
|
return super().__eq__(other)
|
|
|
|
def __ge__(self, other):
|
|
if not isinstance(other, self.__class__):
|
|
other = self.__class__(other)
|
|
return super().__ge__(other)
|
|
|
|
def __gt__(self, other):
|
|
if not isinstance(other, self.__class__):
|
|
other = self.__class__(other)
|
|
return super().__gt__(other)
|
|
|
|
def __ne__(self, other):
|
|
if not isinstance(other, self.__class__):
|
|
other = self.__class__(other)
|
|
return super().__ne__(other)
|
|
|
|
def __hash__(self):
|
|
return hash(str(self))
|