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 = None #: Fingerprint of upload signer

38 # suites

39 suites = [] #: Destination suites

40 from_policy_suites = [] #: Policy suites

42 # package

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

44 changes_filename = None #: Changes Filename

45 sourceful = None #: Did upload contain source

46 source = None #: Source value from changes

47 architecture = None #: Architectures from changes

48 version = None #: Version from changes

49 bugs = None #: Bugs closed in upload

51 # program

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

54 warnings = [] #: Eventual warnings for upload

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

58 """Prepare substitutions used for announce mails.

60 :param upload: upload to handle

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

62 """

63 cnf = Config()

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

66 changed_by = upload.changed_by or maintainer

67 if upload.sourceful:

68 maintainer_to = mail_addresses_for_upload(

69 maintainer, changed_by, upload.fingerprint

70 )

71 else:

72 maintainer_to = mail_addresses_for_upload(

73 maintainer, maintainer, upload.fingerprint

74 )

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

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

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

80 subst = {

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

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

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

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

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

86 "__MANUAL_REJECT_MESSAGE__": "",

87 "__BCC__": bcc,

88 "__MAINTAINER__": changed_by,

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

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

91 "__CHANGES_FILENAME__": upload.changes_filename,

92 "__FILE_CONTENTS__": upload.changes,

93 "__SOURCE__": upload.source,

94 "__VERSION__": upload.version,

95 "__ARCHITECTURE__": upload.architecture,

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

97 }

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

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

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

102

103 return subst

104

105

106def _whitelists(upload: ProcessedUpload):

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

108

109

110def announce_reject(

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

112) -> None:

113 """Announce a reject.

114

115 :param upload: upload to handle

116 :param reason: Reject reason

117 :param rejected_by: Who is doing the reject.

118 """

119 cnf = Config()

120 subst = _subst_for_upload(upload)

121 whitelists = _whitelists(upload)

122

123 automatic = rejected_by is None

124

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

126 "automatic" if automatic else "manual"

127 )

128 subst["__REJECT_MESSAGE__"] = reason

129

130 if rejected_by:

131 subst["__REJECTOR_ADDRESS__"] = rejected_by

132

133 if not automatic:

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

135 subst["__BCC__"], subst["__REJECTOR_ADDRESS__"]

136 )

137

138 message = TemplateSubst(

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

140 )

141 send_mail(message, whitelists=whitelists)

142

143

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

145 """Announce an upload.

146

147 :param upload: upload to handle

148 """

149

150 cnf = Config()

151 subst = _subst_for_upload(upload)

152 whitelists = _whitelists(upload)

153

154 accepted_to_real_suite = any(

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

156 for suite in upload.suites

157 )

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

159

160 suite_names = []

161 for suite in upload.suites:

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

163 suite_names.append(

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

165 )

166 else:

167 suite_names.append(suite.suite_name)

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

169

170 message = TemplateSubst(

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

172 )

173 send_mail(message, whitelists=whitelists)

174

175 if accepted_to_real_suite and upload.sourceful:

176 # send mail to announce lists and tracking server

177 announce = set()

178 for suite in upload.suites:

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

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

181

182 announce_list_address = ", ".join(announce)

183

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

185 # bouncing emails

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

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

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

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

190 announce_list_address, tracker

191 )

192

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

194 my_subst = subst.copy()

195 my_subst["__ANNOUNCE_LIST_ADDRESS__"] = announce_list_address

196

197 message = TemplateSubst(

198 my_subst,

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

200 )

201 send_mail(message, whitelists=whitelists)

202

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

204 close_bugs = any(

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

206 for s in upload.suites

207 )

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