← Back to branch summary

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

« back to all changes in this revision

Viewing changes to pqm_submit.py

  • Committer: James Henstridge
  • Date: 2008-03-25 05:08:50 UTC
  • Revision ID: james@jamesh.id.au-20080325050850-cev83yy11mbrowrj
* Use the standard NoPublicBranch exception to complain about a missing 
  public_branch rather than BzrCommandError.
* Add a NoPQMSubmissionAddress exception for the case when pqm_email is 
  missing from the config file.

Show diffs side-by-side

added added

removed removed

Lines of Context:
pqm_submit.py
 
1
# Copyright (C) 2005, 2006, 2007 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
 
14
# along with this program; if not, write to the Free Software
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  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
        branch_url = urlutils.unescape_for_display(branch.base, 'ascii')
 
46
        errors.BzrError.__init__(self, branch_url=branch_url)
 
47
 
 
48
 
 
49
class PQMEmailMessage(EmailMessage):
 
50
    """PQM doesn't support proper email subjects, so we hack around it."""
 
51
 
 
52
    def __init__(self, from_address, to_address, subject, body=None):
 
53
        EmailMessage.__init__(self, from_address=from_address,
 
54
                              to_address=to_address, subject=subject,
 
55
                              body=body)
 
56
        # Now override self.Subject to use raw utf-8
 
57
        self._headers['Subject'] = osutils.safe_unicode(subject).encode('UTF-8')
 
58
 
 
59
 
 
60
class PQMSubmission(object):
 
61
    """A request to perform a PQM merge into a branch."""
 
62
 
 
63
    def __init__(self, source_branch, public_location=None,
 
64
                 submit_location=None, message=None,
 
65
                 tree=None):
 
66
        """Create a PQMSubmission object.
 
67
 
 
68
        :param source_branch: the source branch for the merge
 
69
        :param public_location: the public location of the source branch
 
70
        :param submit_location: the location of the target branch
 
71
        :param message: The message to use when committing this merge
 
72
        :param tree: A WorkingTree or None. If not None the WT will be checked
 
73
            for uncommitted changes.
 
74
 
 
75
        If any of public_location, submit_location or message are
 
76
        omitted, they will be calculated from source_branch.
 
77
        """
 
78
        if source_branch is None:
 
79
            raise errors.NoMergeSource()
 
80
        self.source_branch = source_branch
 
81
        self.tree = tree
 
82
 
 
83
        if public_location is None:
 
84
            public_location = self.source_branch.get_public_branch()
 
85
            # Fall back to the old public_repository hack.
 
86
            if public_location is None:
 
87
                src_loc = source_branch.bzrdir.root_transport.local_abspath('.')
 
88
                repository = source_branch.repository
 
89
                repo_loc = repository.bzrdir.root_transport.local_abspath('.')
 
90
                repo_config = _mod_config.LocationConfig(repo_loc)
 
91
                public_repo = repo_config.get_user_option("public_repository")
 
92
                if public_repo is not None:
 
93
                    warning("Please use public_branch, not public_repository, "
 
94
                            "to set the public location of branches.")
 
95
                    branch_relpath = osutils.relpath(repo_loc, src_loc)
 
96
                    public_location = urlutils.join(public_repo, branch_relpath)
 
97
 
 
98
            if public_location is None:
 
99
                raise errors.NoPublicBranch(self.source_branch)
 
100
        self.public_location = public_location
 
101
 
 
102
        if submit_location is None:
 
103
            config = self.source_branch.get_config()
 
104
            # First check the deprecated pqm_branch config key:
 
105
            submit_location = config.get_user_option('pqm_branch')
 
106
            if submit_location is not None:
 
107
                warning("Please use submit_branch, not pqm_branch to set "
 
108
                        "the PQM merge target branch.")
 
109
            else:
 
110
                # Otherwise, use the standard config key:
 
111
                submit_location = self.source_branch.get_submit_branch()
 
112
 
 
113
            if submit_location is None:
 
114
                raise errors.NoSubmitBranch(self.source_branch)
 
115
            # See if the submit_location has a public branch
 
116
            try:
 
117
                submit_branch = Branch.open(submit_location)
 
118
            except errors.NotBranchError:
 
119
                pass
 
120
            else:
 
121
                submit_public_location = submit_branch.get_public_branch()
 
122
                if submit_public_location is not None:
 
123
                    submit_location = submit_public_location
 
124
        self.submit_location = submit_location
 
125
 
 
126
        # Check that the message is okay to pass to PQM
 
127
        if message is None:
 
128
            repository = self.source_branch.repository
 
129
            rev = repository.get_revision(self.source_branch.last_revision())
 
130
            message = rev.message
 
131
        self.message = message.encode('utf8')
 
132
        if '\n' in self.message:
 
133
            raise BadCommitMessage(self.message)
 
134
 
 
135
    def check_tree(self):
 
136
        """Check that the working tree has no uncommitted changes."""
 
137
        if self.tree is None:
 
138
            return
 
139
        note('Checking the working tree is clean ...')
 
140
        self.tree.lock_read()
 
141
        try:
 
142
            basis_tree = self.tree.basis_tree()
 
143
            basis_tree.lock_read()
 
144
            try:
 
145
                for change in self.tree.iter_changes(basis_tree):
 
146
                    # If we have any changes, the tree is not clean
 
147
                    raise errors.UncommittedChanges(self.tree)
 
148
            finally:
 
149
                basis_tree.unlock()
 
150
        finally:
 
151
            self.tree.unlock()
 
152
 
 
153
    def check_public_branch(self):
 
154
        """Check that the public branch is up to date with the local copy."""
 
155
        note('Checking that the public branch is up to date ...')
 
156
        local_revision = self.source_branch.last_revision()
 
157
        public_revision = Branch.open(self.public_location).last_revision()
 
158
        if local_revision != public_revision:
 
159
            raise errors.PublicBranchOutOfDate(
 
160
                self.public_location, local_revision)
 
161
 
 
162
    def to_lines(self):
 
163
        """Serialise as a list of lines."""
 
164
        return ['star-merge %s %s\n' % (self.public_location, self.submit_location)]
 
165
 
 
166
    def to_signed(self):
 
167
        """Serialize as a signed string."""
 
168
        unsigned_text = ''.join(self.to_lines())
 
169
        unsigned_text = unsigned_text.encode('ascii') #URLs should be ascii
 
170
 
 
171
        strategy = gpg.GPGStrategy(self.source_branch.get_config())
 
172
        return strategy.sign(unsigned_text)
 
173
 
 
174
    def to_email(self, mail_from, mail_to, sign=True):
 
175
        """Serialize as an email message.
 
176
 
 
177
        :param mail_from: The from address for the message
 
178
        :param mail_to: The address to send the message to
 
179
        :param sign: If True, gpg-sign the email
 
180
        :return: an email message
 
181
        """
 
182
        if sign:
 
183
            body = self.to_signed()
 
184
        else:
 
185
            body = ''.join(self.to_lines())
 
186
        message = PQMEmailMessage(mail_from, mail_to, self.message, body)
 
187
        return message
 
188
 
 
189
 
 
190
def submit(branch, message, dry_run=False, public_location=None,
 
191
           submit_location=None, tree=None):
 
192
    """Submit the given branch to the pqm."""
 
193
    config = branch.get_config()
 
194
 
 
195
    submission = PQMSubmission(
 
196
        source_branch=branch, public_location=public_location, message=message,
 
197
        submit_location=submit_location,
 
198
        tree=tree)
 
199
 
 
200
    mail_from = config.get_user_option('pqm_user_email')
 
201
    if not mail_from:
 
202
        mail_from = config.username()
 
203
    mail_from = mail_from.encode('utf8') # Make sure this isn't unicode
 
204
    mail_to = config.get_user_option('pqm_email')
 
205
    if not mail_to:
 
206
        raise NoPQMSubmissionAddress(branch)
 
207
    mail_to = mail_to.encode('utf8') # same here
 
208
 
 
209
    submission.check_tree()
 
210
    submission.check_public_branch()
 
211
 
 
212
    message = submission.to_email(mail_from, mail_to)
 
213
 
 
214
    if dry_run:
 
215
        print message.as_string()
 
216
        return
 
217
 
 
218
    SMTPConnection(config).send_email(message)