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, Union

from daklib.config import Config
from daklib.textutils import fix_maintainer
from daklib.utils import mail_addresses_for_upload, TemplateSubst, 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)