"""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: Optional[str] = None #: Fingerprint of upload signer
authorized_by_fingerprint: Optional[str] = (
None #: Fingerprint that authorized the upload
)
# 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, upload.authorized_by_fingerprint
)
else:
maintainer_to = mail_addresses_for_upload(
maintainer, maintainer, upload.fingerprint, upload.authorized_by_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)