pungi/pungi/notifier.py
Ozan Unsal ecb1646042 Fix the wrong working directory for the progress_notification script
Jira: RHELCMP-7901

Signed-off-by: Ozan Unsal <ounsal@redhat.com>
2022-03-01 11:25:06 +00:00

113 lines
4.2 KiB
Python

# -*- coding: utf-8 -*-
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; version 2 of the License.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Library General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, see <https://gnu.org/licenses/>.
from datetime import datetime
import json
import os
import threading
import pungi.util
from kobo import shortcuts
class PungiNotifier(object):
"""Wrapper around an external script for sending messages.
If no script is configured, the messages are just silently ignored. If the
script fails, a warning will be logged, but the compose process will not be
interrupted.
"""
def __init__(self, cmds):
self.cmds = cmds
self.lock = threading.Lock()
self.compose = None
def _update_args(self, data):
"""Add compose related information to the data."""
if not self.compose:
return
data.setdefault("compose_id", self.compose.compose_id)
# Publish where in the world this compose will end up living
location = pungi.util.translate_path(
self.compose, self.compose.paths.compose.topdir()
)
data.setdefault("location", location)
# Add information about the compose itself.
data.setdefault("compose_date", self.compose.compose_date)
data.setdefault("compose_type", self.compose.compose_type)
data.setdefault("compose_respin", self.compose.compose_respin)
data.setdefault("compose_label", self.compose.compose_label)
data.setdefault("compose_path", self.compose.topdir)
data.setdefault("release_short", self.compose.conf["release_short"])
data.setdefault("release_name", self.compose.conf["release_name"])
data.setdefault("release_version", self.compose.conf["release_version"])
data.setdefault("release_type", self.compose.conf["release_type"].lower())
data.setdefault("release_is_layered", False)
if self.compose.conf.get("base_product_name", ""):
data["release_is_layered"] = True
data["base_product_name"] = self.compose.conf["base_product_name"]
data["base_product_version"] = self.compose.conf["base_product_version"]
data["base_product_short"] = self.compose.conf["base_product_short"]
data["base_product_type"] = self.compose.conf["base_product_type"].lower()
def send(self, msg, workdir=None, **kwargs):
"""Send a message.
The actual meaning of ``msg`` depends on what the notification script
will be doing. The keyword arguments will be JSON-encoded and passed on
to standard input of the notification process.
Unless you specify it manually, a ``compose_id`` key with appropriate
value will be automatically added.
"""
if not self.cmds:
return
self._update_args(kwargs)
with self.lock:
for cmd in self.cmds:
self._run_script(cmd, msg, workdir, kwargs)
def _run_script(self, cmd, msg, workdir, kwargs):
"""Run a single notification script with proper logging."""
logfile = None
if self.compose:
self.compose.log_debug("Notification: %r %r, %r" % (cmd, msg, kwargs))
logfile = os.path.join(
self.compose.paths.log.topdir(),
"notifications",
"notification-%s.log" % datetime.utcnow().strftime("%Y-%m-%d_%H-%M-%S"),
)
pungi.util.makedirs(os.path.dirname(logfile))
ret, _ = shortcuts.run(
(cmd, msg),
stdin_data=json.dumps(kwargs),
can_fail=True,
workdir=workdir,
return_stdout=False,
show_cmd=True,
universal_newlines=True,
logfile=logfile,
)
if ret != 0:
if self.compose:
self.compose.log_warning("Failed to invoke notification script.")