← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/plugins/launchpad/lp_api.py

  • Committer: John Arbash Meinel
  • Date: 2008-07-09 21:42:24 UTC
  • mto: This revision was merged to the branch mainline in revision 3543.
  • Revision ID: john@arbash-meinel.com-20080709214224-r75k87r6a01pfc3h
Restore a real weave merge to 'bzr merge --weave'.

To do so efficiently, we only add the simple LCAs to the final weave
object, unless we run into complexities with the merge graph.
This gives the same effective result as adding all the texts,
with the advantage of not having to extract all of them.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/plugins/launchpad/lp_api.py
1
 
# Copyright (C) 2009-2012 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., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
 
 
17
 
"""Tools for dealing with the Launchpad API."""
18
 
 
19
 
from __future__ import absolute_import
20
 
 
21
 
# Importing this module will be expensive, since it imports launchpadlib and
22
 
# its dependencies. However, our plan is to only load this module when it is
23
 
# needed by a command that uses it.
24
 
 
25
 
 
26
 
import httplib2
27
 
import os
28
 
import re
29
 
import urlparse
30
 
 
31
 
from bzrlib import (
32
 
    branch,
33
 
    config,
34
 
    errors,
35
 
    osutils,
36
 
    trace,
37
 
    transport,
38
 
    )
39
 
from bzrlib.i18n import gettext
40
 
from bzrlib.plugins.launchpad.lp_registration import (
41
 
    InvalidLaunchpadInstance,
42
 
    )
43
 
 
44
 
try:
45
 
    import launchpadlib
46
 
except ImportError, e:
47
 
    raise errors.DependencyNotPresent('launchpadlib', e)
48
 
 
49
 
from launchpadlib.launchpad import (
50
 
    STAGING_SERVICE_ROOT,
51
 
    Launchpad,
52
 
    )
53
 
from launchpadlib import uris
54
 
 
55
 
# Declare the minimum version of launchpadlib that we need in order to work.
56
 
# 1.6.0 is the version of launchpadlib packaged in Ubuntu 10.04, the most
57
 
# recent Ubuntu LTS release supported on the desktop at the time of writing.
58
 
MINIMUM_LAUNCHPADLIB_VERSION = (1, 6, 0)
59
 
 
60
 
 
61
 
def get_cache_directory():
62
 
    """Return the directory to cache launchpadlib objects in."""
63
 
    return osutils.pathjoin(config.config_dir(), 'launchpad')
64
 
 
65
 
 
66
 
def parse_launchpadlib_version(version_number):
67
 
    """Parse a version number of the style used by launchpadlib."""
68
 
    return tuple(map(int, version_number.split('.')))
69
 
 
70
 
 
71
 
def check_launchpadlib_compatibility():
72
 
    """Raise an error if launchpadlib has the wrong version number."""
73
 
    installed_version = parse_launchpadlib_version(launchpadlib.__version__)
74
 
    if installed_version < MINIMUM_LAUNCHPADLIB_VERSION:
75
 
        raise errors.IncompatibleAPI(
76
 
            'launchpadlib', MINIMUM_LAUNCHPADLIB_VERSION,
77
 
            installed_version, installed_version)
78
 
 
79
 
 
80
 
def lookup_service_root(service_root):
81
 
    try:
82
 
        return uris.lookup_service_root(service_root)
83
 
    except ValueError:
84
 
        if service_root != 'qastaging':
85
 
            raise
86
 
        staging_root = uris.lookup_service_root('staging')
87
 
        return staging_root.replace('staging', 'qastaging')
88
 
 
89
 
 
90
 
def _get_api_url(service):
91
 
    """Return the root URL of the Launchpad API.
92
 
 
93
 
    e.g. For the 'staging' Launchpad service, this function returns
94
 
    launchpadlib.launchpad.STAGING_SERVICE_ROOT.
95
 
 
96
 
    :param service: A `LaunchpadService` object.
97
 
    :return: A URL as a string.
98
 
    """
99
 
    if service._lp_instance is None:
100
 
        lp_instance = service.DEFAULT_INSTANCE
101
 
    else:
102
 
        lp_instance = service._lp_instance
103
 
    try:
104
 
        return lookup_service_root(lp_instance)
105
 
    except ValueError:
106
 
        raise InvalidLaunchpadInstance(lp_instance)
107
 
 
108
 
 
109
 
class NoLaunchpadBranch(errors.BzrError):
110
 
    _fmt = 'No launchpad branch could be found for branch "%(url)s".'
111
 
 
112
 
    def __init__(self, branch):
113
 
        errors.BzrError.__init__(self, branch=branch, url=branch.base)
114
 
 
115
 
 
116
 
def login(service, timeout=None, proxy_info=None,
117
 
          version=Launchpad.DEFAULT_VERSION):
118
 
    """Log in to the Launchpad API.
119
 
 
120
 
    :return: The root `Launchpad` object from launchpadlib.
121
 
    """
122
 
    if proxy_info is None:
123
 
        proxy_info = httplib2.proxy_info_from_environment('https')
124
 
    cache_directory = get_cache_directory()
125
 
    launchpad = Launchpad.login_with(
126
 
        'bzr', _get_api_url(service), cache_directory, timeout=timeout,
127
 
        proxy_info=proxy_info, version=version)
128
 
    # XXX: Work-around a minor security bug in launchpadlib < 1.6.3, which
129
 
    # would create this directory with default umask.
130
 
    osutils.chmod_if_possible(cache_directory, 0700)
131
 
    return launchpad
132
 
 
133
 
 
134
 
class LaunchpadBranch(object):
135
 
    """Provide bzr and lp API access to a Launchpad branch."""
136
 
 
137
 
    def __init__(self, lp_branch, bzr_url, bzr_branch=None, check_update=True):
138
 
        """Constructor.
139
 
 
140
 
        :param lp_branch: The Launchpad branch.
141
 
        :param bzr_url: The URL of the Bazaar branch.
142
 
        :param bzr_branch: An instance of the Bazaar branch.
143
 
        """
144
 
        self.bzr_url = bzr_url
145
 
        self._bzr = bzr_branch
146
 
        self._push_bzr = None
147
 
        self._check_update = check_update
148
 
        self.lp = lp_branch
149
 
 
150
 
    @property
151
 
    def bzr(self):
152
 
        """Return the bzr branch for this branch."""
153
 
        if self._bzr is None:
154
 
            self._bzr = branch.Branch.open(self.bzr_url)
155
 
        return self._bzr
156
 
 
157
 
    @property
158
 
    def push_bzr(self):
159
 
        """Return the push branch for this branch."""
160
 
        if self._push_bzr is None:
161
 
            self._push_bzr = branch.Branch.open(self.lp.bzr_identity)
162
 
        return self._push_bzr
163
 
 
164
 
    @staticmethod
165
 
    def plausible_launchpad_url(url):
166
 
        """Is 'url' something that could conceivably be pushed to LP?
167
 
 
168
 
        :param url: A URL that may refer to a Launchpad branch.
169
 
        :return: A boolean.
170
 
        """
171
 
        if url is None:
172
 
            return False
173
 
        if url.startswith('lp:'):
174
 
            return True
175
 
        regex = re.compile('([a-z]*\+)*(bzr\+ssh|http)'
176
 
                           '://bazaar.*.launchpad.net')
177
 
        return bool(regex.match(url))
178
 
 
179
 
    @staticmethod
180
 
    def candidate_urls(bzr_branch):
181
 
        """Iterate through related URLs that might be Launchpad URLs.
182
 
 
183
 
        :param bzr_branch: A Bazaar branch to find URLs from.
184
 
        :return: a generator of URL strings.
185
 
        """
186
 
        url = bzr_branch.get_public_branch()
187
 
        if url is not None:
188
 
            yield url
189
 
        url = bzr_branch.get_push_location()
190
 
        if url is not None:
191
 
            yield url
192
 
        url = bzr_branch.get_parent()
193
 
        if url is not None:
194
 
            yield url
195
 
        yield bzr_branch.base
196
 
 
197
 
    @staticmethod
198
 
    def tweak_url(url, launchpad):
199
 
        """Adjust a URL to work with staging, if needed."""
200
 
        if str(launchpad._root_uri) == STAGING_SERVICE_ROOT:
201
 
            return url.replace('bazaar.launchpad.net',
202
 
                               'bazaar.staging.launchpad.net')
203
 
        elif str(launchpad._root_uri) == lookup_service_root('qastaging'):
204
 
            return url.replace('bazaar.launchpad.net',
205
 
                               'bazaar.qastaging.launchpad.net')
206
 
        return url
207
 
 
208
 
    @classmethod
209
 
    def from_bzr(cls, launchpad, bzr_branch, create_missing=True):
210
 
        """Find a Launchpad branch from a bzr branch."""
211
 
        check_update = True
212
 
        for url in cls.candidate_urls(bzr_branch):
213
 
            url = cls.tweak_url(url, launchpad)
214
 
            if not cls.plausible_launchpad_url(url):
215
 
                continue
216
 
            lp_branch = launchpad.branches.getByUrl(url=url)
217
 
            if lp_branch is not None:
218
 
                break
219
 
        else:
220
 
            if not create_missing:
221
 
                raise NoLaunchpadBranch(bzr_branch)
222
 
            lp_branch = cls.create_now(launchpad, bzr_branch)
223
 
            check_update = False
224
 
        return cls(lp_branch, bzr_branch.base, bzr_branch, check_update)
225
 
 
226
 
    @classmethod
227
 
    def create_now(cls, launchpad, bzr_branch):
228
 
        """Create a Bazaar branch on Launchpad for the supplied branch."""
229
 
        url = cls.tweak_url(bzr_branch.get_push_location(), launchpad)
230
 
        if not cls.plausible_launchpad_url(url):
231
 
            raise errors.BzrError(gettext('%s is not registered on Launchpad') %
232
 
                                  bzr_branch.base)
233
 
        bzr_branch.create_clone_on_transport(transport.get_transport(url))
234
 
        lp_branch = launchpad.branches.getByUrl(url=url)
235
 
        if lp_branch is None:
236
 
            raise errors.BzrError(gettext('%s is not registered on Launchpad') %
237
 
                                                                            url)
238
 
        return lp_branch
239
 
 
240
 
    def get_target(self):
241
 
        """Return the 'LaunchpadBranch' for the target of this one."""
242
 
        lp_branch = self.lp
243
 
        if lp_branch.project is not None:
244
 
            dev_focus = lp_branch.project.development_focus
245
 
            if dev_focus is None:
246
 
                raise errors.BzrError(gettext('%s has no development focus.') %
247
 
                                  lp_branch.bzr_identity)
248
 
            target = dev_focus.branch
249
 
            if target is None:
250
 
                raise errors.BzrError(gettext(
251
 
                        'development focus %s has no branch.') % dev_focus)
252
 
        elif lp_branch.sourcepackage is not None:
253
 
            target = lp_branch.sourcepackage.getBranch(pocket="Release")
254
 
            if target is None:
255
 
                raise errors.BzrError(gettext(
256
 
                                      'source package %s has no branch.') %
257
 
                                      lp_branch.sourcepackage)
258
 
        else:
259
 
            raise errors.BzrError(gettext(
260
 
                        '%s has no associated product or source package.') %
261
 
                                  lp_branch.bzr_identity)
262
 
        return LaunchpadBranch(target, target.bzr_identity)
263
 
 
264
 
    def update_lp(self):
265
 
        """Update the Launchpad copy of this branch."""
266
 
        if not self._check_update:
267
 
            return
268
 
        self.bzr.lock_read()
269
 
        try:
270
 
            if self.lp.last_scanned_id is not None:
271
 
                if self.bzr.last_revision() == self.lp.last_scanned_id:
272
 
                    trace.note(gettext('%s is already up-to-date.') %
273
 
                               self.lp.bzr_identity)
274
 
                    return
275
 
                graph = self.bzr.repository.get_graph()
276
 
                if not graph.is_ancestor(self.lp.last_scanned_id,
277
 
                                         self.bzr.last_revision()):
278
 
                    raise errors.DivergedBranches(self.bzr, self.push_bzr)
279
 
                trace.note(gettext('Pushing to %s') % self.lp.bzr_identity)
280
 
            self.bzr.push(self.push_bzr)
281
 
        finally:
282
 
            self.bzr.unlock()
283
 
 
284
 
    def find_lca_tree(self, other):
285
 
        """Find the revision tree for the LCA of this branch and other.
286
 
 
287
 
        :param other: Another LaunchpadBranch
288
 
        :return: The RevisionTree of the LCA of this branch and other.
289
 
        """
290
 
        graph = self.bzr.repository.get_graph(other.bzr.repository)
291
 
        lca = graph.find_unique_lca(self.bzr.last_revision(),
292
 
                                    other.bzr.last_revision())
293
 
        return self.bzr.repository.revision_tree(lca)
294
 
 
295
 
 
296
 
def canonical_url(object):
297
 
    """Return the canonical URL for a branch."""
298
 
    scheme, netloc, path, params, query, fragment = urlparse.urlparse(
299
 
        str(object.self_link))
300
 
    path = '/'.join(path.split('/')[2:])
301
 
    netloc = netloc.replace('api.', 'code.')
302
 
    return urlparse.urlunparse((scheme, netloc, path, params, query,
303
 
                                fragment))