Coverage for daklib/announce.py: 84%

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 Optional

25from daklib.config import Config

26from daklib.textutils import fix_maintainer

27from daklib.utils import TemplateSubst, mail_addresses_for_upload, send_mail

30class ProcessedUpload:

31 """Contains data of a processed upload."""

33 # people

34 maintainer = None #: Maintainer: field contents

35 changed_by = None #: Changed-By: field contents

36 fingerprint: Optional[str] = None #: Fingerprint of upload signer

37 authorized_by_fingerprint: Optional[str] = (

38 None #: Fingerprint that authorized the upload

39 )

41 # suites

42 suites = [] #: Destination suites

43 from_policy_suites = [] #: Policy suites

45 # package

46 changes = None #: Contents of .changes file from upload

47 changes_filename = None #: Changes Filename

48 sourceful = None #: Did upload contain source

49 source = None #: Source value from changes

50 architecture = None #: Architectures from changes

51 version = None #: Version from changes

52 bugs = None #: Bugs closed in upload

54 # program

55 program = "unknown-program" #: Which dak program was in use

57 warnings = [] #: Eventual warnings for upload

60def _subst_for_upload(upload: ProcessedUpload) -> dict:

61 """Prepare substitutions used for announce mails.

63 :param upload: upload to handle

64 :return: A dict of substition values for use by :func:`daklib.utils.TemplateSubst`

65 """

66 cnf = Config()

68 maintainer = upload.maintainer or cnf["Dinstall::MyEmailAddress"]

69 changed_by = upload.changed_by or maintainer

70 if upload.sourceful:

71 maintainer_to = mail_addresses_for_upload(

72 maintainer, changed_by, upload.fingerprint, upload.authorized_by_fingerprint

73 )

74 else:

75 maintainer_to = mail_addresses_for_upload(

76 maintainer, maintainer, upload.fingerprint, upload.authorized_by_fingerprint

77 )

79 bcc = "X-DAK: dak {0}".format(upload.program)

80 if "Dinstall::Bcc" in cnf: 80 ↛ 81line 80 didn't jump to line 81, because the condition on line 80 was never true

81 bcc = "{0}\nBcc: {1}".format(bcc, cnf["Dinstall::Bcc"])

83 subst = {

84 "__DISTRO__": cnf["Dinstall::MyDistribution"],

85 "__BUG_SERVER__": cnf.get("Dinstall::BugServer"),

86 "__ADMIN_ADDRESS__": cnf["Dinstall::MyAdminAddress"],

87 "__DAK_ADDRESS__": cnf["Dinstall::MyEmailAddress"],

88 "__REJECTOR_ADDRESS__": cnf["Dinstall::MyEmailAddress"],

89 "__MANUAL_REJECT_MESSAGE__": "",

90 "__BCC__": bcc,

91 "__MAINTAINER__": changed_by,

92 "__MAINTAINER_FROM__": fix_maintainer(changed_by)[1],

93 "__MAINTAINER_TO__": ", ".join(maintainer_to),

94 "__CHANGES_FILENAME__": upload.changes_filename,

95 "__FILE_CONTENTS__": upload.changes,

96 "__SOURCE__": upload.source,

97 "__VERSION__": upload.version,

98 "__ARCHITECTURE__": upload.architecture,

99 "__WARNINGS__": "\n".join(upload.warnings),

100 }

101

102 override_maintainer = cnf.get("Dinstall::OverrideMaintainer")

103 if override_maintainer: 103 ↛ 104line 103 didn't jump to line 104, because the condition on line 103 was never true

104 subst["__MAINTAINER_FROM__"] = subst["__MAINTAINER_TO__"] = override_maintainer

105

106 return subst

107

108

109def _whitelists(upload: ProcessedUpload):

110 return [s.mail_whitelist for s in upload.suites]

111

112

113def announce_reject(

114 upload: ProcessedUpload, reason: str, rejected_by: Optional[str] = None

115) -> None:

116 """Announce a reject.

117

118 :param upload: upload to handle

119 :param reason: Reject reason

120 :param rejected_by: Who is doing the reject.

121 """

122 cnf = Config()

123 subst = _subst_for_upload(upload)

124 whitelists = _whitelists(upload)

125

126 automatic = rejected_by is None

127

128 subst["__CC__"] = "X-DAK-Rejection: {0}".format(

129 "automatic" if automatic else "manual"

130 )

131 subst["__REJECT_MESSAGE__"] = reason

132

133 if rejected_by:

134 subst["__REJECTOR_ADDRESS__"] = rejected_by

135

136 if not automatic:

137 subst["__BCC__"] = "{0}\nBcc: {1}".format(

138 subst["__BCC__"], subst["__REJECTOR_ADDRESS__"]

139 )

140

141 message = TemplateSubst(

142 subst, os.path.join(cnf["Dir::Templates"], "queue.rejected")

143 )

144 send_mail(message, whitelists=whitelists)

145

146

147def announce_accept(upload: ProcessedUpload) -> None:

148 """Announce an upload.

149

150 :param upload: upload to handle

151 """

152

153 cnf = Config()

154 subst = _subst_for_upload(upload)

155 whitelists = _whitelists(upload)

156

157 accepted_to_real_suite = any(

158 suite.policy_queue is None or suite in upload.from_policy_suites

159 for suite in upload.suites

160 )

161 subst["__ACTION__"] = "accept" if accepted_to_real_suite else "policy"

162

163 suite_names = []

164 for suite in upload.suites:

165 if suite.policy_queue and suite not in upload.from_policy_suites:

166 suite_names.append(

167 "{0}->{1}".format(suite.suite_name, suite.policy_queue.queue_name)

168 )

169 else:

170 suite_names.append(suite.suite_name)

171 subst["__SUITE__"] = ", ".join(suite_names) or "(none)"

172

173 message = TemplateSubst(

174 subst, os.path.join(cnf["Dir::Templates"], "process-unchecked.accepted")

175 )

176 send_mail(message, whitelists=whitelists)

177

178 if accepted_to_real_suite and upload.sourceful:

179 # send mail to announce lists and tracking server

180 announce = set()

181 for suite in upload.suites:

182 if suite.policy_queue is None or suite in upload.from_policy_suites: 182 ↛ 181line 182 didn't jump to line 181, because the condition on line 182 was never false

183 announce.update(suite.announce or [])

184

185 announce_list_address = ", ".join(announce)

186

187 # according to #890944 this email shall be sent to dispatch@<TrackingServer> to avoid

188 # bouncing emails

189 # the package email alias is not yet created shortly after accepting the package

190 tracker = cnf.get("Dinstall::TrackingServer")

191 if tracker: 191 ↛ 192line 191 didn't jump to line 192, because the condition on line 191 was never true

192 announce_list_address = "{0}\nBcc: dispatch@{1}".format(

193 announce_list_address, tracker

194 )

195

196 if len(announce_list_address) != 0: 196 ↛ 197line 196 didn't jump to line 197, because the condition on line 196 was never true

197 my_subst = subst.copy()

198 my_subst["__ANNOUNCE_LIST_ADDRESS__"] = announce_list_address

199

200 message = TemplateSubst(

201 my_subst,

202 os.path.join(cnf["Dir::Templates"], "process-unchecked.announce"),

203 )

204 send_mail(message, whitelists=whitelists)

205

206 close_bugs_default = cnf.find_b("Dinstall::CloseBugs")

207 close_bugs = any(

208 s.close_bugs if s.close_bugs is not None else close_bugs_default

209 for s in upload.suites

210 )

211 if accepted_to_real_suite and upload.sourceful and close_bugs: 211 ↛ 212line 211 didn't jump to line 212, because the condition on line 211 was never true