1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
|
# Copyright 2009 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""Email notifications related to branch merge proposals."""
__metaclass__ = type
from canonical.config import config
from canonical.launchpad.mail import get_msgid
from canonical.launchpad.webapp import canonical_url
from lp.code.enums import CodeReviewNotificationLevel
from lp.code.mail.branch import BranchMailer
from lp.services.mail.basemailer import BaseMailer
class BMPMailer(BranchMailer):
"""Send mailings related to BranchMergeProposal events."""
def __init__(self, subject, template_name, recipients, merge_proposal,
from_address, delta=None, message_id=None,
requested_reviews=None, preview_diff=None,
direct_email=False):
BranchMailer.__init__(
self, subject, template_name, recipients, from_address,
message_id=message_id, notification_type='code-review')
self.merge_proposal = merge_proposal
if requested_reviews is None:
requested_reviews = []
self.requested_reviews = requested_reviews
self.preview_diff = preview_diff
self.delta_text = delta
self.template_params = self._generateTemplateParams()
self.direct_email = direct_email
def sendAll(self):
BranchMailer.sendAll(self)
if self.merge_proposal.root_message_id is None:
self.merge_proposal.root_message_id = self.message_id
@classmethod
def forCreation(cls, merge_proposal, from_user):
"""Return a mailer for BranchMergeProposal creation.
:param merge_proposal: The BranchMergeProposal that was created.
:param from_user: The user that the creation notification should
come from.
"""
recipients = merge_proposal.getNotificationRecipients(
CodeReviewNotificationLevel.STATUS)
assert from_user.preferredemail is not None, (
'The sender must have an email address.')
from_address = cls._format_user_address(from_user)
return cls(
'%(proposal_title)s',
'branch-merge-proposal-created.txt', recipients, merge_proposal,
from_address, message_id=get_msgid(),
requested_reviews=merge_proposal.votes,
preview_diff=merge_proposal.preview_diff)
@classmethod
def forModification(cls, merge_proposal, delta_text, from_user):
"""Return a mailer for BranchMergeProposal creation.
:param merge_proposal: The BranchMergeProposal that was created.
:param from_user: The user that the creation notification should
come from. Optional.
"""
recipients = merge_proposal.getNotificationRecipients(
CodeReviewNotificationLevel.STATUS)
if from_user is not None:
assert from_user.preferredemail is not None, (
'The sender must have an email address.')
from_address = cls._format_user_address(from_user)
else:
from_address = config.canonical.noreply_from_address
return cls(
'%(proposal_title)s',
'branch-merge-proposal-updated.txt', recipients,
merge_proposal, from_address, delta=delta_text,
message_id=get_msgid())
@classmethod
def forReviewRequest(cls, reason, merge_proposal, from_user):
"""Return a mailer for a request to review a BranchMergeProposal."""
from_address = cls._format_user_address(from_user)
recipients = {reason.subscriber: reason}
return cls(
'%(proposal_title)s',
'review-requested.txt', recipients,
merge_proposal, from_address, message_id=get_msgid(),
preview_diff=merge_proposal.preview_diff, direct_email=True)
def _getReplyToAddress(self):
"""Return the address to use for the reply-to header."""
return self.merge_proposal.address
def _getToAddresses(self, recipient, email):
"""Return the addresses to use for the to header.
If the email is being sent directly to the recipient, their email
address is returned. Otherwise, the merge proposal and requested
reviewers are returned.
"""
if self.direct_email:
return BaseMailer._getToAddresses(self, recipient, email)
to_addrs = [self.merge_proposal.address]
for vote in self.merge_proposal.votes:
if vote.reviewer == vote.registrant:
continue
if vote.reviewer.is_team:
continue
if vote.reviewer.hide_email_addresses:
continue
to_addrs.append(self._format_user_address(vote.reviewer))
return to_addrs
def _getHeaders(self, email):
"""Return the mail headers to use."""
headers = BranchMailer._getHeaders(self, email)
if self.merge_proposal.root_message_id is not None:
headers['In-Reply-To'] = self.merge_proposal.root_message_id
return headers
def _addAttachments(self, ctrl, email):
if self.preview_diff is not None:
reason, rationale = self._recipients.getReason(email)
if reason.review_level == CodeReviewNotificationLevel.FULL:
# Using .txt as a file extension makes Gmail display it
# inline.
ctrl.addAttachment(
self.preview_diff.text, content_type='text/x-diff',
inline=True, filename='review-diff.txt')
def _generateTemplateParams(self):
"""For template params that don't change, calculate just once."""
proposal = self.merge_proposal
params = {
'proposal_registrant': proposal.registrant.displayname,
'source_branch': proposal.source_branch.bzr_identity,
'target_branch': proposal.target_branch.bzr_identity,
'prerequisite': '',
'proposal_title': proposal.title,
'proposal_url': canonical_url(proposal),
'edit_subscription': '',
'comment': '',
'gap': '',
'reviews': '',
'whiteboard': '', # No more whiteboard.
'diff_cutoff_warning': '',
}
if self.delta_text is not None:
params['delta'] = self.delta_text
if proposal.prerequisite_branch is not None:
prereq_url = proposal.prerequisite_branch.bzr_identity
params['prerequisite'] = ' with %s as a prerequisite' % prereq_url
requested_reviews = []
for review in self.requested_reviews:
reviewer = review.reviewer
if review.review_type is None:
requested_reviews.append(reviewer.unique_displayname)
else:
requested_reviews.append(
"%s: %s" % (reviewer.unique_displayname,
review.review_type))
if len(requested_reviews) > 0:
requested_reviews.insert(0, 'Requested reviews:')
params['reviews'] = (''.join(' %s\n' % review
for review in requested_reviews))
if proposal.description is not None:
params['comment'] = (proposal.description)
if len(requested_reviews) > 0:
params['gap'] = '\n\n'
if (self.preview_diff is not None and self.preview_diff.oversized):
params['diff_cutoff_warning'] = (
"The attached diff has been truncated due to its size.\n")
params['reviews'] = self._getRequestedReviews()
return params
def _formatExtraInformation(self, heading, chunks):
"""Consistently indent the chunks with the heading.
Used to provide consistent indentation for requested reviews and
related bugs.
"""
if len(chunks) == 0:
return ''
else:
info = ''.join(' %s\n' % value for value in chunks)
return '%s\n%s' % (heading, info)
def _getRequestedReviews(self):
"""Return a string describing the requested reviews, if any."""
requested_reviews = []
for review in self.requested_reviews:
reviewer = review.reviewer
if review.review_type is None:
requested_reviews.append(reviewer.unique_displayname)
else:
requested_reviews.append(
"%s: %s" % (reviewer.unique_displayname,
review.review_type))
return self._formatExtraInformation(
'Requested reviews:', requested_reviews)
def _getRelatedBugTasks(self, recipient):
"""Return a string describing related bug tasks, if any.
Related bugs are provided by
`IBranchMergeProposal.getRelatedBugTasks`
"""
bug_chunks = []
for bugtask in self.merge_proposal.getRelatedBugTasks(recipient):
bug_chunks.append('%s' % bugtask.title)
bug_chunks.append(canonical_url(bugtask))
return self._formatExtraInformation('Related bugs:', bug_chunks)
def _getTemplateParams(self, email, recipient):
"""Return a dict of values to use in the body and subject."""
# Expand the requested reviews.
params = BranchMailer._getTemplateParams(self, email, recipient)
params['related_bugtasks'] = self._getRelatedBugTasks(recipient)
params.update(self.template_params)
return params
|