~ursinha/lp-qa-tools/bzr-tarmacland

« back to all changes in this revision

Viewing changes to pqm_submit.py

  • Committer: Ursula Junque (Ursinha)
  • Date: 2010-12-16 03:14:26 UTC
  • Revision ID: ursinha@canonical.com-20101216031426-bbk2xmnt9vs8hvqz
Removing unnecessary files for the tarmac-land command.

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005-2008 by Canonical Ltd
2
 
#
3
 
# This program is free software; you can redistribute it and/or modify
4
 
# it under the terms of the GNU General Public License as published by
5
 
# the Free Software Foundation; either version 2 of the License, or
6
 
# (at your option) any later version.
7
 
#
8
 
# This program is distributed in the hope that it will be useful,
9
 
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
 
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
11
 
# GNU General Public License for more details.
12
 
#
13
 
# You should have received a copy of the GNU General Public License along
14
 
# with this program; if not, write to the Free Software Foundation, Inc.,
15
 
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
16
 
"""Submit an email to a Patch Queue Manager"""
17
 
 
18
 
from bzrlib import (
19
 
    config as _mod_config,
20
 
    errors,
21
 
    gpg,
22
 
    osutils,
23
 
    urlutils,
24
 
    )
25
 
from bzrlib.branch import Branch
26
 
from bzrlib.email_message import EmailMessage
27
 
from bzrlib.smtp_connection import SMTPConnection
28
 
from bzrlib.trace import note, warning
29
 
 
30
 
 
31
 
class BadCommitMessage(errors.BzrError):
32
 
 
33
 
    _fmt = "The commit message %(msg)r cannot be used by pqm."
34
 
 
35
 
    def __init__(self, message):
36
 
        errors.BzrError.__init__(self)
37
 
        self.msg = message
38
 
 
39
 
 
40
 
class NoPQMSubmissionAddress(errors.BzrError):
41
 
 
42
 
    _fmt = "No PQM submission email address specified for %(branch_url)s."
43
 
 
44
 
    def __init__(self, branch):
45
 
        if (branch is None) or (branch.base is None):
46
 
            branch_url = '(none)'
47
 
        else:
48
 
            branch_url = urlutils.unescape_for_display(branch.base, 'ascii')
49
 
        errors.BzrError.__init__(self, branch_url=branch_url)
50
 
 
51
 
 
52
 
class PQMEmailMessage(EmailMessage):
53
 
    """PQM doesn't support proper email subjects, so we hack around it."""
54
 
 
55
 
    def __init__(self, from_address, to_address, subject, body=None):
56
 
        EmailMessage.__init__(self, from_address=from_address,
57
 
                              to_address=to_address, subject=subject,
58
 
                              body=body)
59
 
        # Now override self.Subject to use raw utf-8
60
 
        self._headers['Subject'] = osutils.safe_unicode(subject).encode('UTF-8')
61
 
 
62
 
 
63
 
class PQMSubmission(object):
64
 
    """A request to perform a PQM merge into a branch."""
65
 
 
66
 
    def __init__(self, source_branch, public_location=None,
67
 
                 submit_location=None, message=None,
68
 
                 tree=None):
69
 
        """Create a PQMSubmission object.
70
 
 
71
 
        :param source_branch: the source branch for the merge
72
 
        :param public_location: the public location of the source branch
73
 
        :param submit_location: the location of the target branch
74
 
        :param message: The message to use when committing this merge
75
 
        :param tree: A WorkingTree or None. If not None the WT will be checked
76
 
            for uncommitted changes.
77
 
 
78
 
        If any of public_location, submit_location or message are
79
 
        omitted, they will be calculated from source_branch.
80
 
        """
81
 
        if source_branch is None and public_location is None:
82
 
            raise errors.NoMergeSource()
83
 
        self.source_branch = source_branch
84
 
        self.tree = tree
85
 
 
86
 
        if public_location is None:
87
 
            public_location = self.source_branch.get_public_branch()
88
 
            # Fall back to the old public_repository hack.
89
 
            if public_location is None:
90
 
                src_loc = source_branch.bzrdir.root_transport.local_abspath('.')
91
 
                repository = source_branch.repository
92
 
                repo_loc = repository.bzrdir.root_transport.local_abspath('.')
93
 
                repo_config = _mod_config.LocationConfig(repo_loc)
94
 
                public_repo = repo_config.get_user_option("public_repository")
95
 
                if public_repo is not None:
96
 
                    warning("Please use public_branch, not public_repository, "
97
 
                            "to set the public location of branches.")
98
 
                    branch_relpath = osutils.relpath(repo_loc, src_loc)
99
 
                    public_location = urlutils.join(public_repo, branch_relpath)
100
 
 
101
 
            if public_location is None:
102
 
                raise errors.NoPublicBranch(self.source_branch)
103
 
        self.public_location = public_location
104
 
 
105
 
        if submit_location is None:
106
 
            if self.source_branch is None:
107
 
                raise errors.BzrError(
108
 
                    "Cannot determine submit location to use.")
109
 
            config = self.source_branch.get_config()
110
 
            # First check the deprecated pqm_branch config key:
111
 
            submit_location = config.get_user_option('pqm_branch')
112
 
            if submit_location is not None:
113
 
                warning("Please use submit_branch, not pqm_branch to set "
114
 
                        "the PQM merge target branch.")
115
 
            else:
116
 
                # Otherwise, use the standard config key:
117
 
                submit_location = self.source_branch.get_submit_branch()
118
 
 
119
 
            if submit_location is None:
120
 
                raise errors.NoSubmitBranch(self.source_branch)
121
 
            # See if the submit_location has a public branch
122
 
            try:
123
 
                submit_branch = Branch.open(submit_location)
124
 
            except errors.NotBranchError:
125
 
                pass
126
 
            else:
127
 
                submit_public_location = submit_branch.get_public_branch()
128
 
                if submit_public_location is not None:
129
 
                    submit_location = submit_public_location
130
 
        self.submit_location = submit_location
131
 
 
132
 
        # Check that the message is okay to pass to PQM
133
 
        assert message is not None
134
 
        self.message = message.encode('utf8')
135
 
        if '\n' in self.message:
136
 
            raise BadCommitMessage(self.message)
137
 
 
138
 
    def check_tree(self):
139
 
        """Check that the working tree has no uncommitted changes."""
140
 
        if self.tree is None:
141
 
            return
142
 
        note('Checking the working tree is clean ...')
143
 
        self.tree.lock_read()
144
 
        try:
145
 
            basis_tree = self.tree.basis_tree()
146
 
            basis_tree.lock_read()
147
 
            try:
148
 
                for change in self.tree.iter_changes(basis_tree):
149
 
                    # If we have any changes, the tree is not clean
150
 
                    raise errors.UncommittedChanges(self.tree)
151
 
            finally:
152
 
                basis_tree.unlock()
153
 
        finally:
154
 
            self.tree.unlock()
155
 
 
156
 
    def check_public_branch(self):
157
 
        """Check that the public branch is up to date with the local copy."""
158
 
        note('Checking that the public branch is up to date at\n    %s',
159
 
             urlutils.unescape_for_display(self.public_location, 'utf-8'))
160
 
        local_revision = self.source_branch.last_revision()
161
 
        public_revision = Branch.open(self.public_location).last_revision()
162
 
        if local_revision != public_revision:
163
 
            raise errors.PublicBranchOutOfDate(
164
 
                self.public_location, local_revision)
165
 
 
166
 
    def to_lines(self):
167
 
        """Serialise as a list of lines."""
168
 
        return ['star-merge %s %s\n' % (self.public_location, self.submit_location)]
169
 
 
170
 
    def to_signed(self):
171
 
        """Serialize as a signed string."""
172
 
        unsigned_text = ''.join(self.to_lines())
173
 
        unsigned_text = unsigned_text.encode('ascii') #URLs should be ascii
174
 
 
175
 
        if self.source_branch:
176
 
            config = self.source_branch.get_config()
177
 
        else:
178
 
            config = _mod_config.GlobalConfig()
179
 
        strategy = gpg.GPGStrategy(config)
180
 
        return strategy.sign(unsigned_text)
181
 
 
182
 
    def to_email(self, mail_from, mail_to, sign=True):
183
 
        """Serialize as an email message.
184
 
 
185
 
        :param mail_from: The from address for the message
186
 
        :param mail_to: The address to send the message to
187
 
        :param sign: If True, gpg-sign the email
188
 
        :return: an email message
189
 
        """
190
 
        if sign:
191
 
            body = self.to_signed()
192
 
        else:
193
 
            body = ''.join(self.to_lines())
194
 
        message = PQMEmailMessage(mail_from, mail_to, self.message, body)
195
 
        return message
196
 
 
197
 
 
198
 
class StackedConfig(_mod_config.Config):
199
 
 
200
 
    def __init__(self):
201
 
        super(StackedConfig, self).__init__()
202
 
        self._sources = []
203
 
 
204
 
    def add_source(self, source):
205
 
        self._sources.append(source)
206
 
 
207
 
    def _get_user_option(self, option_name):
208
 
        """See Config._get_user_option."""
209
 
        for source in self._sources:
210
 
            value = source._get_user_option(option_name)
211
 
            if value is not None:
212
 
                return value
213
 
        return None
214
 
 
215
 
    def _get_user_id(self):
216
 
        for source in self._sources:
217
 
            value = source._get_user_id()
218
 
            if value is not None:
219
 
                return value
220
 
        return None
221
 
 
222
 
 
223
 
def submit(branch, message, dry_run=False, public_location=None,
224
 
           submit_location=None, tree=None, ignore_local=False):
225
 
    """Submit the given branch to the pqm."""
226
 
    config = StackedConfig()
227
 
    if branch:
228
 
        config.add_source(branch.get_config())
229
 
    else:
230
 
        if public_location:
231
 
            config.add_source(_mod_config.LocationConfig(public_location))
232
 
        config.add_source(_mod_config.GlobalConfig())
233
 
 
234
 
    submission = PQMSubmission(
235
 
        source_branch=branch, public_location=public_location, message=message,
236
 
        submit_location=submit_location,
237
 
        tree=tree)
238
 
 
239
 
    mail_from = config.get_user_option('pqm_user_email')
240
 
    if not mail_from:
241
 
        mail_from = config.username()
242
 
    mail_from = mail_from.encode('utf8') # Make sure this isn't unicode
243
 
    mail_to = config.get_user_option('pqm_email')
244
 
    if not mail_to:
245
 
        submit_branch = Branch.open(submit_location)
246
 
        submit_branch_config = submit_branch.get_config()
247
 
        mail_to = submit_branch_config.get_user_option('child_pqm_email')
248
 
        if not mail_to:
249
 
            raise NoPQMSubmissionAddress(branch)
250
 
    mail_to = mail_to.encode('utf8') # same here
251
 
 
252
 
    if not ignore_local:
253
 
        submission.check_tree()
254
 
        submission.check_public_branch()
255
 
 
256
 
    message = submission.to_email(mail_from, mail_to)
257
 
 
258
 
    mail_bcc = config.get_user_option('pqm_bcc')
259
 
    if mail_bcc is not None:
260
 
        message["Bcc"] = mail_bcc
261
 
 
262
 
    if dry_run:
263
 
        print message.as_string()
264
 
        return
265
 
 
266
 
    SMTPConnection(config).send_email(message)