chore(*): cleanup buildbot-effects

Signed-off-by: Raito Bezarius <raito@lix.systems>
This commit is contained in:
raito 2024-05-06 18:26:01 +02:00 committed by Jade Lovelace
parent 7875db31eb
commit 58bc2cddae
5 changed files with 0 additions and 352 deletions

View file

@ -1,9 +0,0 @@
#!/usr/bin/env python
import sys
from pathlib import Path
sys.path.append(str(Path(__file__).parent.parent))
from hercules_effects.cli import main
if __name__ == '__main__':
main()

View file

@ -1,243 +0,0 @@
import json
import os
import shlex
import shutil
import subprocess
import sys
from collections.abc import Iterator
from contextlib import contextmanager
from pathlib import Path
from tempfile import NamedTemporaryFile
from typing import IO, Any
from .options import EffectsOptions
class BuildbotEffectsError(Exception):
pass
def run(
cmd: list[str],
stdin: int | IO[str] | None = None,
stdout: int | IO[str] | None = None,
stderr: int | IO[str] | None = None,
verbose: bool = True,
) -> subprocess.CompletedProcess[str]:
if verbose:
print("$", shlex.join(cmd), file=sys.stderr)
return subprocess.run(
cmd,
check=True,
text=True,
stdin=stdin,
stdout=stdout,
stderr=stderr,
)
def git_command(args: list[str], path: Path) -> str:
cmd = ["git", "-C", str(path), *args]
proc = run(cmd, stdout=subprocess.PIPE)
return proc.stdout.strip()
def get_git_rev(path: Path) -> str:
return git_command(["rev-parse", "--verify", "HEAD"], path)
def get_git_branch(path: Path) -> str:
return git_command(["rev-parse", "--abbrev-ref", "HEAD"], path)
def get_git_remote_url(path: Path) -> str | None:
try:
return git_command(["remote", "get-url", "origin"], path)
except subprocess.CalledProcessError:
return None
def git_get_tag(path: Path, rev: str) -> str | None:
tags = git_command(["tag", "--points-at", rev], path)
if tags:
return tags.splitlines()[1]
return None
def effects_args(opts: EffectsOptions) -> dict[str, Any]:
rev = opts.rev or get_git_rev(opts.path)
short_rev = rev[:7]
branch = opts.branch or get_git_branch(opts.path)
repo = opts.repo or opts.path.name
tag = opts.tag or git_get_tag(opts.path, rev)
url = opts.url or get_git_remote_url(opts.path)
primary_repo = dict(
name=repo,
branch=branch,
# TODO: support ref
ref=None,
tag=tag,
rev=rev,
shortRev=short_rev,
remoteHttpUrl=url,
)
return {
"primaryRepo": primary_repo,
**primary_repo,
}
def nix_command(*args: str) -> list[str]:
return ["nix", "--extra-experimental-features", "nix-command flakes", *args]
def effect_function(opts: EffectsOptions) -> str:
args = effects_args(opts)
rev = args["rev"]
escaped_args = json.dumps(json.dumps(args))
url = json.dumps(f"git+file://{opts.path}?rev={rev}#")
return f"""(((builtins.getFlake {url}).outputs.herculesCI (builtins.fromJSON {escaped_args})).onPush.default.outputs.hci-effects)"""
def list_effects(opts: EffectsOptions) -> list[str]:
cmd = nix_command(
"eval",
"--json",
"--expr",
f"builtins.attrNames {effect_function(opts)}",
)
proc = run(cmd, stdout=subprocess.PIPE)
return json.loads(proc.stdout)
def instantiate_effects(opts: EffectsOptions) -> str:
cmd = [
"nix-instantiate",
"--expr",
f"{effect_function(opts)}.deploy.run",
]
proc = run(cmd, stdout=subprocess.PIPE)
return proc.stdout.rstrip()
def parse_derivation(path: str) -> dict[str, Any]:
cmd = [
"nix",
"--extra-experimental-features",
"nix-command flakes",
"derivation",
"show",
f"{path}^*",
]
proc = run(cmd, stdout=subprocess.PIPE)
return json.loads(proc.stdout)
def env_args(env: dict[str, str]) -> list[str]:
result = []
for k, v in env.items():
result.append("--setenv")
result.append(f"{k}")
result.append(f"{v}")
return result
@contextmanager
def pipe() -> Iterator[tuple[IO[str], IO[str]]]:
r, w = os.pipe()
r_file = os.fdopen(r, "r")
w_file = os.fdopen(w, "w")
try:
yield r_file, w_file
finally:
r_file.close()
w_file.close()
def run_effects(
drv_path: str,
drv: dict[str, Any],
secrets: dict[str, Any] | None = None,
) -> None:
if secrets is None:
secrets = {}
builder = drv["builder"]
args = drv["args"]
sandboxed_cmd = [
builder,
*args,
]
env = {}
env["IN_HERCULES_CI_EFFECT"] = "true"
env["HERCULES_CI_SECRETS_JSON"] = "/run/secrets.json"
env["NIX_BUILD_TOP"] = "/build"
bwrap = shutil.which("bwrap")
if bwrap is None:
msg = "bwrap' executable not found"
raise BuildbotEffectsError(msg)
bubblewrap_cmd = [
"nix",
"develop",
"-i",
f"{drv_path}^*",
"-c",
bwrap,
"--unshare-all",
"--share-net",
"--new-session",
"--die-with-parent",
"--dir",
"/build",
"--chdir",
"/build",
"--tmpfs",
"/tmp", # noqa: S108
"--tmpfs",
"/build",
"--proc",
"/proc",
"--dev",
"/dev",
"--ro-bind",
"/etc/resolv.conf",
"/etc/resolv.conf",
"--ro-bind",
"/etc/hosts",
"/etc/hosts",
"--ro-bind",
"/nix/store",
"/nix/store",
]
with NamedTemporaryFile() as tmp:
secrets = secrets.copy()
secrets["hercules-ci"] = {"data": {"token": "dummy"}}
tmp.write(json.dumps(secrets).encode())
bubblewrap_cmd.extend(
[
"--ro-bind",
tmp.name,
"/run/secrets.json",
],
)
bubblewrap_cmd.extend(env_args(env))
bubblewrap_cmd.append("--")
bubblewrap_cmd.extend(sandboxed_cmd)
with pipe() as (r_file, w_file):
print("$", shlex.join(bubblewrap_cmd), file=sys.stderr)
proc = subprocess.Popen(
bubblewrap_cmd,
text=True,
stdin=subprocess.DEVNULL,
stdout=w_file,
stderr=w_file,
)
w_file.close()
with proc:
for line in r_file:
print(line, end="")
proc.wait()
if proc.returncode != 0:
msg = f"command failed with exit code {proc.returncode}"
raise BuildbotEffectsError(msg)

View file

@ -1,85 +0,0 @@
import argparse
import json
from collections.abc import Callable
from pathlib import Path
from . import instantiate_effects, list_effects, parse_derivation, run_effects
from .options import EffectsOptions
def list_command(options: EffectsOptions) -> None:
print(list_effects(options))
def run_command(options: EffectsOptions) -> None:
drv_path = instantiate_effects(options)
drvs = parse_derivation(drv_path)
drv = next(iter(drvs.values()))
secrets = json.loads(options.secrets.read_text()) if options.secrets else {}
run_effects(drv_path, drv, secrets=secrets)
def run_all_command(options: EffectsOptions) -> None:
print("TODO")
def parse_args() -> tuple[Callable[[EffectsOptions], None], EffectsOptions]:
parser = argparse.ArgumentParser(description="Run effects from a hercules-ci flake")
parser.add_argument(
"--secrets",
type=Path,
help="Path to a json file with secrets",
)
parser.add_argument(
"--rev",
type=str,
help="Git revision to use",
)
parser.add_argument(
"--branch",
type=str,
help="Git branch to use",
)
parser.add_argument(
"--repo",
type=str,
help="Git repo to prepend to be",
)
parser.add_argument(
"--path",
type=str,
help="Path to the repository",
)
subparser = parser.add_subparsers(
dest="command",
required=True,
help="Command to run",
)
list_parser = subparser.add_parser(
"list",
help="List available effects",
)
list_parser.set_defaults(command=list_command)
run_parser = subparser.add_parser(
"run",
help="Run an effect",
)
run_parser.set_defaults(command=run_command)
run_parser.add_argument(
"effect",
help="Effect to run",
)
run_all_parser = subparser.add_parser(
"run-all",
help="Run all effects",
)
run_all_parser.set_defaults(command=run_all_command)
args = parser.parse_args()
return args.command, EffectsOptions(secrets=args.secrets)
def main() -> None:
command, options = parse_args()
command(options)

View file

@ -1,13 +0,0 @@
from dataclasses import dataclass, field
from pathlib import Path
@dataclass
class EffectsOptions:
secrets: Path | None = None
path: Path = field(default_factory=lambda: Path.cwd())
repo: str | None = ""
rev: str | None = None
branch: str | None = None
url: str | None = None
tag: str | None = None

View file

@ -21,12 +21,10 @@ classifiers = [
"Programming Language :: Python" "Programming Language :: Python"
] ]
version = "0.0.1" version = "0.0.1"
scripts = { buildbot-effects = "hercules_effects.cli:main" }
[tool.setuptools] [tool.setuptools]
packages = [ packages = [
"buildbot_nix", "buildbot_nix",
"buildbot_effects"
] ]
[tool.ruff] [tool.ruff]