from buildbot.plugins import util, steps, schedulers
from buildbot_buildslist import BuildsList
from shutil import which
__all__ = [
"force_scheduler", "deploy_scheduler", "git_hook_scheduler",
"clean_branch", "package_and_upload", "AppriseStatusPush",
"XMPPStatusPush", "NixShellCommand",
"all_builder_names", "compute_build_infos", "deploy_ssh_command",
"configure_apprise_push",
"configure_xmpp_push", "deploy_hook_scheduler",
]
# Small helpers"
@util.renderer
def clean_branch(props):
if props.hasProperty("branch") and len(props["branch"]) > 0:
return props["branch"].replace("/", "_")
else:
return "HEAD"
def package_and_upload(package, package_dest, package_url):
return [
steps.ShellCommand(name="build package",
logEnviron=False, haltOnFailure=True,
command=["git", "archive", "HEAD", "-o", package]),
steps.FileUpload(name="upload package", workersrc=package,
masterdest=package_dest,
url=package_url, mode=0o644),
steps.ShellCommand(name="cleanup package", logEnviron=False,
haltOnFailure=True, alwaysRun=True,
command=["rm", "-f", package]),
]
# Steps
class NixShellCommand(steps.ShellCommand):
def __init__(self, command=None, nixPackages=[], pure=True, nixFile=None, nixIncludes={}, nixArgs={}, **kwargs):
oldpath = kwargs.get("env", {}).get("PATH", None)
if which("nix-shell", path=oldpath) is None:
kwargs["env"] = kwargs.get("env", {})
if isinstance(oldpath, str):
kwargs["env"]["PATH"] = "/run/current-system/sw/bin:" + oldpath
elif isinstance(oldpath, list):
kwargs["env"]["PATH"] = ["/run/current-system/sw/bin"] + oldpath
nixcommand = ["nix-shell"]
for k, v in nixArgs.items():
nixcommand.append("--arg")
nixcommand.append(k)
nixcommand.append(v)
if pure:
nixcommand.append("--pure")
for k, v in nixIncludes.items():
nixcommand.append("-I")
nixcommand.append("{}={}".format(k, v))
nixcommand.append("--run")
nixcommand.append(command)
if len(nixPackages) > 0:
nixcommand.append("-p")
nixcommand += nixPackages
elif nixFile is not None:
nixcommand.append(nixFile)
super().__init__(command=nixcommand, **kwargs)
# Schedulers
def force_scheduler(name, builders, nobranch=False):
if nobranch:
branch = util.FixedParameter(name="branch", default="")
else:
branch=util.StringParameter(name="branch", label="Git reference (tag, branch)", required=True)
return schedulers.ForceScheduler(name=name,
label="Force build", buttonName="Force build",
reason=util.StringParameter(name="reason", label="Reason", default="Force build"),
codebases=[
util.CodebaseParameter("",
branch=branch,
revision=util.FixedParameter(name="revision", default=""),
repository=util.FixedParameter(name="repository", default=""),
project=util.FixedParameter(name="project", default=""),
),
],
username=util.FixedParameter(name="username", default="Web button"),
builderNames=builders)
def deploy_scheduler(name, builders):
return schedulers.ForceScheduler(name=name,
builderNames=builders,
label="Deploy built package", buttonName="Deploy",
username=util.FixedParameter(name="username", default="Web button"),
codebases=[
util.CodebaseParameter(codebase="",
branch=util.FixedParameter(name="branch", default=""),
revision=util.FixedParameter(name="revision", default=""),
repository=util.FixedParameter(name="repository", default=""),
project=util.FixedParameter(name="project", default=""))],
reason=util.FixedParameter(name="reason", default="Deploy"),
properties=[
util.ChoiceStringParameter(label="Environment",
name="environment", default="integration",
choices=["integration", "production"]),
BuildsList(label="Build to deploy", name="build"),
]
)
def git_hook_scheduler(project, builders=[], timer=1):
if len(builders) == 0:
builders = ["{}_build".format(project)]
return schedulers.AnyBranchScheduler(
change_filter=util.ChangeFilter(category="gitolite-hooks", project=project),
name="{}_git_hook".format(project), treeStableTimer=timer, builderNames=builders)
def deploy_hook_scheduler(project, builders, timer=1):
return schedulers.AnyBranchScheduler(
change_filter=util.ChangeFilter(category="deploy_webhook", project=project),
name="{}_deploy".format(project), treeStableTimer=timer, builderNames=builders)
# Builders
def all_builder_names(c):
return [builder.name for builder in c['builders']]
# Apprise/XMPP status push
from buildbot.reporters.http import HttpStatusPushBase
from twisted.internet import defer
from twisted.python import log
from buildbot.reporters import utils
from buildbot.process import results
from twisted.words.protocols.jabber.jid import JID
from wokkel import client, xmppim
from functools import partial
import apprise
class AppriseStatusPush(HttpStatusPushBase):
name = "AppriseStatusPush"
@defer.inlineCallbacks
def reconfigService(self, appriseUrls, **kwargs):
self.appriseUrls = appriseUrls
yield HttpStatusPushBase.reconfigService(self, **kwargs)
@defer.inlineCallbacks
def send(self, build):
yield utils.getDetailsForBuild(self.master, build, wantProperties=True)
appobject = apprise.Apprise()
message = self.format(build)
for url in self.appriseUrls:
appobject.add(url.format(**message))
yield appobject.notify(title=message["title"], body=message["text"])
def format(self, build):
if "environment" in build["properties"]:
msg = "{} environment".format(build["properties"]["environment"][0])
if "build" in build["properties"]:
msg = "of archive {} in ".format(build["properties"]["build"][0]) + msg
elif len(build["buildset"]["sourcestamps"][0]["branch"] or []) > 0:
msg = "revision {}".format(build["buildset"]["sourcestamps"][0]["branch"])
else:
msg = "build"
if build["complete"]:
timedelta = int((build["complete_at"] - build["started_at"]).total_seconds())
hours, rest = divmod(timedelta, 3600)
minutes, seconds = divmod(rest, 60)
if hours > 0:
duration = "{}h {}min {}s".format(hours, minutes, seconds)
elif minutes > 0:
duration = "{}min {}s".format(minutes, seconds)
else:
duration = "{}s".format(seconds)
text = "Build {} ({}) of {}'s {} was {} in {}.".format(
build["number"], build["url"],
build["builder"]["name"],
msg,
results.Results[build["results"]],
duration,
)
else:
text = "Build {} ({}) of {}'s {} started.".format(
build["number"], build["url"],
build["builder"]["name"],
msg,
)
return {
"username": "Buildbot",
"image_url": "http://docs.buildbot.net/current/_static/icon.png",
"text": text,
"title": "",
}
def configure_apprise_push(c, secrets_file, builders):
c['services'].append(AppriseStatusPush(
name="apprise_status", builders=builders,
appriseUrls=open(secrets_file + "/apprise_webhooks", "r").read().split("\n")))
class XMPPStatusPush(HttpStatusPushBase):
name = "XMPPStatusPush"
@defer.inlineCallbacks
def reconfigService(self, password, recipients, **kwargs):
yield HttpStatusPushBase.reconfigService(self, **kwargs)
self.password = password
self.recipients = recipients
@defer.inlineCallbacks
def send(self, build):
yield utils.getDetailsForBuild(self.master, build, wantProperties=True)
body = self.format(build)
factory = client.DeferredClientFactory(JID("notify_bot@immae.fr/buildbot"), self.password)
d = client.clientCreator(factory)
def send_message(recipient, stream):
message = xmppim.Message(recipient=JID(recipient), body=body)
message.stanzaType = 'chat'
stream.send(message.toElement())
# To allow chaining
return stream
for recipient in self.recipients:
d.addCallback(partial(send_message, recipient))
d.addCallback(lambda _: factory.streamManager.xmlstream.sendFooter())
d.addErrback(log.err)
def format(self, build):
if "environment" in build["properties"]:
msg = "{} environment".format(build["properties"]["environment"][0])
if "build" in build["properties"]:
msg = "of archive {} in ".format(build["properties"]["build"][0]) + msg
elif len(build["buildset"]["sourcestamps"][0]["branch"] or []) > 0:
msg = "revision {}".format(build["buildset"]["sourcestamps"][0]["branch"])
else:
msg = "build"
if build["complete"]:
timedelta = int((build["complete_at"] - build["started_at"]).total_seconds())
hours, rest = divmod(timedelta, 3600)
minutes, seconds = divmod(rest, 60)
if hours > 0:
duration = "{}h {}min {}s".format(hours, minutes, seconds)
elif minutes > 0:
duration = "{}min {}s".format(minutes, seconds)
else:
duration = "{}s".format(seconds)
text = "Build {} ( {} ) of {}'s {} was {} in {}.".format(
build["number"], build["url"],
build["builder"]["name"],
msg,
results.Results[build["results"]],
duration,
)
else:
text = "Build {} ( {} ) of {}'s {} started.".format(
build["number"], build["url"],
build["builder"]["name"],
msg,
)
return text
def configure_xmpp_push(c, secrets_file, builders, recipients):
c['services'].append(XMPPStatusPush(
name="xmpp_status", builders=builders, recipients=recipients,
password=open(secrets_file + "/notify_xmpp_password", "r").read().rstrip()))
# LDAP edit
from buildbot.process.buildstep import FAILURE
from buildbot.process.buildstep import SUCCESS
from buildbot.process.buildstep import BuildStep
def compute_build_infos(prefix, release_path):
@util.renderer
def compute(props):
import re, hashlib
build_file = props.getProperty("build")
package_dest = "{}/{}".format(release_path, build_file)
version = re.match(r"{0}_(.*).tar.gz".format(prefix), build_file).group(1)
with open(package_dest, "rb") as f:
sha = hashlib.sha256(f.read()).hexdigest()
return {
"build_version": version,
"build_hash": sha,
}
return compute
def deploy_ssh_command(ssh_key_path, deploy_hosts):
@util.renderer
def compute(props):
environment = props["environment"] if props.hasProperty("environment") else "integration"
ssh_command = [
"ssh", "-o", "UserKnownHostsFile=/dev/null", "-o", "StrictHostKeyChecking=no", "-o", "CheckHostIP=no",
"-i", ssh_key_path ]
return ssh_command + deploy_hosts.get(environment, ["host.invalid"])
return compute