Coverage for daklib/announce.py: 84%
94 statements
« prev ^ index » next coverage.py v7.6.0, created at 2026-01-04 16:18 +0000
1"""module to send announcements for processed packages
3@contact: Debian FTP Master <ftpmaster@debian.org>
4@copyright: 2012, Ansgar Burchardt <ansgar@debian.org>
5@license: GPL-2+
6"""
8# This program is free software; you can redistribute it and/or modify
9# it under the terms of the GNU General Public License as published by
10# the Free Software Foundation; either version 2 of the License, or
11# (at your option) any later version.
12#
13# This program is distributed in the hope that it will be useful,
14# but WITHOUT ANY WARRANTY; without even the implied warranty of
15# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16# GNU General Public License for more details.
17#
18# You should have received a copy of the GNU General Public License along
19# with this program; if not, write to the Free Software Foundation, Inc.,
20# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
22import os
23from typing import TYPE_CHECKING, Optional
25from daklib.config import Config
26from daklib.textutils import fix_maintainer
27from daklib.utils import TemplateSubst, mail_addresses_for_upload, send_mail
29if TYPE_CHECKING:
30 from daklib.dbconn import Suite
33class ProcessedUpload:
34 """Contains data of a processed upload."""
36 # people
37 maintainer: str | None = None #: Maintainer: field contents
38 changed_by: str | None = None #: Changed-By: field contents
39 fingerprint: Optional[str] = None #: Fingerprint of upload signer
40 authorized_by_fingerprint: Optional[str] = (
41 None #: Fingerprint that authorized the upload
42 )
44 # suites
45 suites: list["Suite"] = [] #: Destination suites
46 from_policy_suites: list["Suite"] = [] #: Policy suites
48 # package
49 changes: str | None = None #: Contents of .changes file from upload
50 changes_filename: str | None = None #: Changes Filename
51 sourceful: bool | None = None #: Did upload contain source
52 source: str | None = None #: Source value from changes
53 architecture: str | None = None #: Architectures from changes
54 version: str | None = None #: Version from changes
55 bugs: list[str] | None = None #: Bugs closed in upload
57 # program
58 program = "unknown-program" #: Which dak program was in use
60 warnings: list[str] = [] #: Eventual warnings for upload
63def _subst_for_upload(upload: ProcessedUpload) -> dict:
64 """Prepare substitutions used for announce mails.
66 :param upload: upload to handle
67 :return: A dict of substition values for use by :func:`daklib.utils.TemplateSubst`
68 """
69 assert upload.fingerprint is not None
71 cnf = Config()
73 maintainer = upload.maintainer or cnf["Dinstall::MyEmailAddress"]
74 changed_by = upload.changed_by or maintainer
75 if upload.sourceful:
76 maintainer_to = mail_addresses_for_upload(
77 maintainer, changed_by, upload.fingerprint, upload.authorized_by_fingerprint
78 )
79 else:
80 maintainer_to = mail_addresses_for_upload(
81 maintainer, maintainer, upload.fingerprint, upload.authorized_by_fingerprint
82 )
84 bcc = "X-DAK: dak {0}".format(upload.program)
85 if "Dinstall::Bcc" in cnf: 85 ↛ 86line 85 didn't jump to line 86 because the condition on line 85 was never true
86 bcc = "{0}\nBcc: {1}".format(bcc, cnf["Dinstall::Bcc"])
88 subst = {
89 "__DISTRO__": cnf["Dinstall::MyDistribution"],
90 "__BUG_SERVER__": cnf.get("Dinstall::BugServer"),
91 "__ADMIN_ADDRESS__": cnf["Dinstall::MyAdminAddress"],
92 "__DAK_ADDRESS__": cnf["Dinstall::MyEmailAddress"],
93 "__REJECTOR_ADDRESS__": cnf["Dinstall::MyEmailAddress"],
94 "__MANUAL_REJECT_MESSAGE__": "",
95 "__BCC__": bcc,
96 "__MAINTAINER__": changed_by,
97 "__MAINTAINER_FROM__": fix_maintainer(changed_by)[1],
98 "__MAINTAINER_TO__": ", ".join(maintainer_to),
99 "__CHANGES_FILENAME__": upload.changes_filename,
100 "__FILE_CONTENTS__": upload.changes,
101 "__SOURCE__": upload.source,
102 "__VERSION__": upload.version,
103 "__ARCHITECTURE__": upload.architecture,
104 "__WARNINGS__": "\n".join(upload.warnings),
105 }
107 override_maintainer = cnf.get("Dinstall::OverrideMaintainer")
108 if override_maintainer: 108 ↛ 109line 108 didn't jump to line 109 because the condition on line 108 was never true
109 subst["__MAINTAINER_FROM__"] = subst["__MAINTAINER_TO__"] = override_maintainer
111 return subst
114def _whitelists(upload: ProcessedUpload) -> list[str | None]:
115 return [s.mail_whitelist for s in upload.suites]
118def announce_reject(
119 upload: ProcessedUpload, reason: str, rejected_by: Optional[str] = None
120) -> None:
121 """Announce a reject.
123 :param upload: upload to handle
124 :param reason: Reject reason
125 :param rejected_by: Who is doing the reject.
126 """
127 cnf = Config()
128 subst = _subst_for_upload(upload)
129 whitelists = _whitelists(upload)
131 automatic = rejected_by is None
133 subst["__CC__"] = "X-DAK-Rejection: {0}".format(
134 "automatic" if automatic else "manual"
135 )
136 subst["__REJECT_MESSAGE__"] = reason
138 if rejected_by:
139 subst["__REJECTOR_ADDRESS__"] = rejected_by
141 if not automatic:
142 subst["__BCC__"] = "{0}\nBcc: {1}".format(
143 subst["__BCC__"], subst["__REJECTOR_ADDRESS__"]
144 )
146 message = TemplateSubst(
147 subst, os.path.join(cnf["Dir::Templates"], "queue.rejected")
148 )
149 send_mail(message, whitelists=whitelists)
152def announce_accept(upload: ProcessedUpload) -> None:
153 """Announce an upload.
155 :param upload: upload to handle
156 """
158 cnf = Config()
159 subst = _subst_for_upload(upload)
160 whitelists = _whitelists(upload)
162 accepted_to_real_suite = any(
163 suite.policy_queue is None or suite in upload.from_policy_suites
164 for suite in upload.suites
165 )
166 subst["__ACTION__"] = "accept" if accepted_to_real_suite else "policy"
168 suite_names = []
169 for suite in upload.suites:
170 if suite.policy_queue and suite not in upload.from_policy_suites:
171 suite_names.append(
172 "{0}->{1}".format(suite.suite_name, suite.policy_queue.queue_name)
173 )
174 else:
175 suite_names.append(suite.suite_name)
176 subst["__SUITE__"] = ", ".join(suite_names) or "(none)"
178 message = TemplateSubst(
179 subst, os.path.join(cnf["Dir::Templates"], "process-unchecked.accepted")
180 )
181 send_mail(message, whitelists=whitelists)
183 if accepted_to_real_suite and upload.sourceful:
184 # send mail to announce lists and tracking server
185 announce = set()
186 for suite in upload.suites:
187 if suite.policy_queue is None or suite in upload.from_policy_suites: 187 ↛ 186line 187 didn't jump to line 186 because the condition on line 187 was always true
188 announce.update(suite.announce or [])
190 announce_list_address = ", ".join(announce)
192 # according to #890944 this email shall be sent to dispatch@<TrackingServer> to avoid
193 # bouncing emails
194 # the package email alias is not yet created shortly after accepting the package
195 tracker = cnf.get("Dinstall::TrackingServer")
196 if tracker: 196 ↛ 197line 196 didn't jump to line 197 because the condition on line 196 was never true
197 announce_list_address = "{0}\nBcc: dispatch@{1}".format(
198 announce_list_address, tracker
199 )
201 if len(announce_list_address) != 0: 201 ↛ 202line 201 didn't jump to line 202 because the condition on line 201 was never true
202 my_subst = subst.copy()
203 my_subst["__ANNOUNCE_LIST_ADDRESS__"] = announce_list_address
205 message = TemplateSubst(
206 my_subst,
207 os.path.join(cnf["Dir::Templates"], "process-unchecked.announce"),
208 )
209 send_mail(message, whitelists=whitelists)
211 close_bugs_default = cnf.find_b("Dinstall::CloseBugs")
212 close_bugs = any(
213 s.close_bugs if s.close_bugs is not None else close_bugs_default
214 for s in upload.suites or []
215 )
216 if accepted_to_real_suite and upload.sourceful and close_bugs: 216 ↛ 217line 216 didn't jump to line 217 because the condition on line 216 was never true
217 for bug in upload.bugs or []:
218 my_subst = subst.copy()
219 my_subst["__BUG_NUMBER__"] = str(bug)
221 message = TemplateSubst(
222 my_subst,
223 os.path.join(cnf["Dir::Templates"], "process-unchecked.bug-close"),
224 )
225 send_mail(message, whitelists=whitelists)
228def announce_new(upload: ProcessedUpload) -> None:
229 """Announce an upload going to NEW.
231 :param upload: upload to handle
232 """
234 cnf = Config()
235 subst = _subst_for_upload(upload)
236 whitelists = _whitelists(upload)
238 message = TemplateSubst(
239 subst, os.path.join(cnf["Dir::Templates"], "process-unchecked.new")
240 )
241 send_mail(message, whitelists=whitelists)