~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

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

Rework test_script a little bit.


Don't allow someone to request a stdin request to echo.
Echo never reads from stdin, it just echos its arguments.
You use 'cat' if you want to read from stdin.

A few other fixes because the tests were using filenames
that are actually illegal on Windows, rather than just
nonexistant.


Change the exception handling for commands so that
unknown errors don't get silently squashed and then
turn into hard-to-debug errors later.

test_script now passes on Windows.

Show diffs side-by-side

added added

removed removed

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