# Copyright (c) 2018 Cisco and/or its affiliates. # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at: # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import smtplib import logging from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from os.path import isdir from errors import PresentationError class AlertingError(PresentationError): """Exception(s) raised by the alerting module. When raising this exception, put this information to the message in this order: - short description of the encountered problem (parameter msg), - relevant messages if there are any collected, e.g., from caught exception (optional parameter details), - relevant data if there are any collected (optional parameter details). """ def __init__(self, msg, details='', level="CRITICAL"): """Sets the exception message and the level. :param msg: Short description of the encountered problem. :param details: Relevant messages if there are any collected, e.g., from caught exception (optional parameter details), or relevant data if there are any collected (optional parameter details). :param level: Level of the error, possible choices are: "DEBUG", "INFO", "WARNING", "ERROR" and "CRITICAL". :type msg: str :type details: str :type level: str """ super(AlertingError, self).__init__( "Alerting: {0}".format(msg), details, level) def __repr__(self): return ( "AlertingError(msg={msg!r},details={dets!r},level={level!r})". format(msg=self._msg, dets=self._details, level=self._level)) class Alerting(object): """Class implementing the alerting mechanism. """ def __init__(self, spec): """Initialization. :param spec: The CPTA specification. :type spec: Specification """ # Implemented alerts: self._ALERTS = ("failed-tests", ) self._spec = spec.alerting self._path_failed_tests = spec.environment["paths"]["DIR[STATIC,VPP]"] # Verify and validate input specification: self.configs = self._spec.get("configurations", None) if not self.configs: raise AlertingError("No alert configuration is specified.") for config_type, config_data in self.configs.iteritems(): if config_type == "email": if not config_data.get("server", None): raise AlertingError("Parameter 'server' is missing.") if not config_data.get("address-to", None): raise AlertingError("Parameter 'address-to' (recipient) is " "missing.") if not config_data.get("address-from", None): raise AlertingError("Parameter 'address-from' (sender) is " "missing.") elif config_type == "jenkins": if not isdir(config_data.get("output-dir", "")): raise AlertingError("Parameter 'output-dir' is " "missing or it is not a directory.") if not config_data.get("output-file", None): raise AlertingError("Parameter 'output-file' is missing.") else: raise AlertingError("Alert of type '{0}' is not implemented.". format(config_type)) self.alerts = self._spec.get("alerts", None) if not self.alerts: raise AlertingError("No alert is specified.") for alert, alert_data in self.alerts.iteritems(): if not alert_data.get("title", None): raise AlertingError("Parameter 'title' is missing.") if not alert_data.get("type", None) in self._ALERTS: raise AlertingError("Parameter 'failed-tests' is missing or " "incorrect.") if not alert_data.get("way", None) in self.configs.keys(): raise AlertingError("Parameter 'way' is missing or incorrect.") if not alert_data.get("include", None): raise AlertingError("Parameter 'include' is missing or the " "list is empty.") def __str__(self): """Return string with human readable description of the alert. :returns: Readable description. :rtype: str """ return "configs={configs}, alerts={alerts}".format( configs=self.configs, alerts=self.alerts) def __repr__(self): """Return string executable as Python constructor call. :returns: Executable constructor call. :rtype: str """ return "Alerting(spec={spec})".format( spec=self._spec) def generate_alerts(self): """Generate alert(s) using specified way(s). """ for alert, alert_data in self.alerts.iteritems(): if alert_data["way"] == "email": text, html = self._create_alert_message(alert_data) conf = self.configs["email"] self._send_email(server=conf["server"], addr_from=conf["address-from"], addr_to=conf["address-to"], subject=alert_data["title"], text=text, html=html) elif alert_data["way"] == "jenkins": self._generate_files_for_jenkins(alert_data) else: raise AlertingError("Alert with way '{0}' is not implemented.". format(alert_data["way"])) @staticmethod def _send_email(server, addr_from, addr_to, subject, text=None, html=None): """Send an email using predefined configuration. :param server: SMTP server used to send email. :param addr_from: Sender address. :param addr_to: Recipient address(es). :param subject: Subject of the email. :param text: Message in the ASCII text format. :param html: Message in the HTML format. :type server: str :type addr_from: str :type addr_to: list :type subject: str :type text: str :type html: str """ if not text and not html: raise AlertingError("No text/data to send.") msg = MIMEMultipart('alternative') msg['Subject'] = subject msg['From'] = addr_from msg['To'] = ", ".join(addr_to) if text: msg.attach(MIMEText(text, 'plain')) if html: msg.attach(MIMEText(html, 'html')) smtp_server = None try: logging.info("Trying to send alert '{0}' ...".format(subject)) logging.debug("SMTP Server: {0}".format(server)) logging.debug("From: {0}".format(addr_from)) logging.debug("To: {0}".format(", ".join(addr_to))) logging.debug("Message: {0}".format(msg.as_string())) smtp_server = smtplib.SMTP(server) smtp_server.sendmail(addr_from, addr_to, msg.as_string()) except smtplib.SMTPException as err: raise AlertingError("Not possible to send the alert via email.", str(err)) finally: if smtp_server: smtp_server.quit() def _create_alert_message(self, alert): """Create the message which is used in the generated alert. :param alert: Message is created for this alert. :type alert: dict :returns: Message in the ASCII text and HTML format. :rtype: tuple(str, str) """ if alert["type"] == "failed-tests": text = "" html = "" for item in alert["include"]: file_name = "{path}/{name}".format( path=self._path_failed_tests, name=item) try: with open("{0}.txt".format(file_name), 'r') as txt_file: text += "{0}:\n\n".format( item.replace("failed-tests-", "")) text += txt_file.read() + "\n" * 2 except IOError: logging.error("Not possible to read the file '{0}.txt'.". format(file_name)) try: with open("{0}.rst".format(file_name), 'r') as rst_file: html += "

{0}:

".format( item.replace("failed-tests-", "")) html += rst_file.readlines()[2].\ replace("../trending", alert.get("url", "")) html += "
" * 3 except IOError: logging.error("Not possible to read the file '{0}.rst'.". format(file_name)) html += "" else: raise AlertingError("Alert of type '{0}' is not implemented.". format(alert["type"])) return text, html def _generate_files_for_jenkins(self, alert): """Create the file which is used in the generated alert. :param alert: Files are created for this alert. :type alert: dict """ config = self.configs[alert["way"]] if alert["type"] == "failed-tests": text, html = self._create_alert_message(alert) file_name = "{0}/{1}".format(config["output-dir"], config["output-file"]) logging.info("Writing the file '{0}.txt' ...".format(file_name)) try: with open("{0}.txt".format(file_name), 'w') as txt_file: txt_file.write(text) except IOError: logging.error("Not possible to write the file '{0}.txt'.". format(file_name)) logging.info("Writing the file '{0}.html' ...".format(file_name)) try: with open("{0}.html".format(file_name), 'w') as html_file: html_file.write(html) except IOError: logging.error("Not possible to write the file '{0}.html'.". format(file_name)) else: raise AlertingError("Alert of type '{0}' is not implemented.". format(alert["type"]))