Source code for daklib.announce

"""module to send announcements for processed packages

@contact: Debian FTP Master <ftpmaster@debian.org>
@copyright: 2012, Ansgar Burchardt <ansgar@debian.org>
@license: GPL-2+
"""

# 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; either version 2 of the License, or
# (at your option) any later version.
#
# 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.

import os
from typing import Optional

from daklib.config import Config
from daklib.textutils import fix_maintainer
from daklib.utils import TemplateSubst, mail_addresses_for_upload, send_mail


[docs]class ProcessedUpload: """Contains data of a processed upload.""" # people maintainer = None #: Maintainer: field contents changed_by = None #: Changed-By: field contents fingerprint = None #: Fingerprint of upload signer # suites suites = [] #: Destination suites from_policy_suites = [] #: Policy suites # package changes = None #: Contents of .changes file from upload changes_filename = None #: Changes Filename sourceful = None #: Did upload contain source source = None #: Source value from changes architecture = None #: Architectures from changes version = None #: Version from changes bugs = None #: Bugs closed in upload # program program = "unknown-program" #: Which dak program was in use warnings = [] #: Eventual warnings for upload
[docs]def _subst_for_upload(upload: ProcessedUpload) -> dict: """Prepare substitutions used for announce mails. :param upload: upload to handle :return: A dict of substition values for use by :func:`daklib.utils.TemplateSubst` """ cnf = Config() maintainer = upload.maintainer or cnf["Dinstall::MyEmailAddress"] changed_by = upload.changed_by or maintainer if upload.sourceful: maintainer_to = mail_addresses_for_upload( maintainer, changed_by, upload.fingerprint ) else: maintainer_to = mail_addresses_for_upload( maintainer, maintainer, upload.fingerprint ) bcc = "X-DAK: dak {0}".format(upload.program) if "Dinstall::Bcc" in cnf: bcc = "{0}\nBcc: {1}".format(bcc, cnf["Dinstall::Bcc"]) subst = { "__DISTRO__": cnf["Dinstall::MyDistribution"], "__BUG_SERVER__": cnf.get("Dinstall::BugServer"), "__ADMIN_ADDRESS__": cnf["Dinstall::MyAdminAddress"], "__DAK_ADDRESS__": cnf["Dinstall::MyEmailAddress"], "__REJECTOR_ADDRESS__": cnf["Dinstall::MyEmailAddress"], "__MANUAL_REJECT_MESSAGE__": "", "__BCC__": bcc, "__MAINTAINER__": changed_by, "__MAINTAINER_FROM__": fix_maintainer(changed_by)[1], "__MAINTAINER_TO__": ", ".join(maintainer_to), "__CHANGES_FILENAME__": upload.changes_filename, "__FILE_CONTENTS__": upload.changes, "__SOURCE__": upload.source, "__VERSION__": upload.version, "__ARCHITECTURE__": upload.architecture, "__WARNINGS__": "\n".join(upload.warnings), } override_maintainer = cnf.get("Dinstall::OverrideMaintainer") if override_maintainer: subst["__MAINTAINER_FROM__"] = subst["__MAINTAINER_TO__"] = override_maintainer return subst
[docs]def _whitelists(upload: ProcessedUpload): return [s.mail_whitelist for s in upload.suites]
[docs]def announce_reject( upload: ProcessedUpload, reason: str, rejected_by: Optional[str] = None ) -> None: """Announce a reject. :param upload: upload to handle :param reason: Reject reason :param rejected_by: Who is doing the reject. """ cnf = Config() subst = _subst_for_upload(upload) whitelists = _whitelists(upload) automatic = rejected_by is None subst["__CC__"] = "X-DAK-Rejection: {0}".format( "automatic" if automatic else "manual" ) subst["__REJECT_MESSAGE__"] = reason if rejected_by: subst["__REJECTOR_ADDRESS__"] = rejected_by if not automatic: subst["__BCC__"] = "{0}\nBcc: {1}".format( subst["__BCC__"], subst["__REJECTOR_ADDRESS__"] ) message = TemplateSubst( subst, os.path.join(cnf["Dir::Templates"], "queue.rejected") ) send_mail(message, whitelists=whitelists)
[docs]def announce_accept(upload: ProcessedUpload) -> None: """Announce an upload. :param upload: upload to handle """ cnf = Config() subst = _subst_for_upload(upload) whitelists = _whitelists(upload) accepted_to_real_suite = any( suite.policy_queue is None or suite in upload.from_policy_suites for suite in upload.suites ) subst["__ACTION__"] = "accept" if accepted_to_real_suite else "policy" suite_names = [] for suite in upload.suites: if suite.policy_queue and suite not in upload.from_policy_suites: suite_names.append( "{0}->{1}".format(suite.suite_name, suite.policy_queue.queue_name) ) else: suite_names.append(suite.suite_name) subst["__SUITE__"] = ", ".join(suite_names) or "(none)" message = TemplateSubst( subst, os.path.join(cnf["Dir::Templates"], "process-unchecked.accepted") ) send_mail(message, whitelists=whitelists) if accepted_to_real_suite and upload.sourceful: # send mail to announce lists and tracking server announce = set() for suite in upload.suites: if suite.policy_queue is None or suite in upload.from_policy_suites: announce.update(suite.announce or []) announce_list_address = ", ".join(announce) # according to #890944 this email shall be sent to dispatch@<TrackingServer> to avoid # bouncing emails # the package email alias is not yet created shortly after accepting the package tracker = cnf.get("Dinstall::TrackingServer") if tracker: announce_list_address = "{0}\nBcc: dispatch@{1}".format( announce_list_address, tracker ) if len(announce_list_address) != 0: my_subst = subst.copy() my_subst["__ANNOUNCE_LIST_ADDRESS__"] = announce_list_address message = TemplateSubst( my_subst, os.path.join(cnf["Dir::Templates"], "process-unchecked.announce"), ) send_mail(message, whitelists=whitelists) close_bugs_default = cnf.find_b("Dinstall::CloseBugs") close_bugs = any( s.close_bugs if s.close_bugs is not None else close_bugs_default for s in upload.suites ) if accepted_to_real_suite and upload.sourceful and close_bugs: for bug in upload.bugs: my_subst = subst.copy() my_subst["__BUG_NUMBER__"] = str(bug) message = TemplateSubst( my_subst, os.path.join(cnf["Dir::Templates"], "process-unchecked.bug-close"), ) send_mail(message, whitelists=whitelists)
[docs]def announce_new(upload: ProcessedUpload) -> None: """Announce an upload going to NEW. :param upload: upload to handle """ cnf = Config() subst = _subst_for_upload(upload) whitelists = _whitelists(upload) message = TemplateSubst( subst, os.path.join(cnf["Dir::Templates"], "process-unchecked.new") ) send_mail(message, whitelists=whitelists)