1
# Copyright (C) 2007 Canonical Ltd
 
 
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.
 
 
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.
 
 
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
 
 
18
"""Transport indirection that uses Launchpad as a directory lookup.
 
 
20
When the transport is opened, it immediately redirects to a url
 
 
21
on Launchpad, which can then either serve the branch itself or redirect
 
 
25
from urlparse import urlsplit, urlunsplit
 
 
34
from bzrlib.transport import (
 
 
36
    register_urlparse_netloc_protocol,
 
 
40
from bzrlib.plugins.launchpad.lp_registration import (
 
 
41
    LaunchpadService, ResolveLaunchpadPathRequest)
 
 
42
from bzrlib.plugins.launchpad.account import get_lp_login
 
 
45
# As bzrlib.transport.remote may not be loaded yet, make sure bzr+ssh
 
 
46
# is counted as a netloc protocol.
 
 
47
register_urlparse_netloc_protocol('bzr+ssh')
 
 
48
register_urlparse_netloc_protocol('lp')
 
 
51
class LaunchpadTransport(Transport):
 
 
52
    """lp:/// URL transport
 
 
54
    This transport redirects requests to the real branch location
 
 
55
    after resolving the URL via an XMLRPC request to Launchpad.
 
 
58
    def __init__(self, base):
 
 
59
        super(LaunchpadTransport, self).__init__(base)
 
 
60
        # We only support URLs without a netloc
 
 
61
        netloc = urlsplit(base)[1]
 
 
63
            raise errors.InvalidURL(path=base)
 
 
65
    def _resolve(self, abspath,
 
 
66
                 _request_factory=ResolveLaunchpadPathRequest,
 
 
68
        """Resolve the base URL for this transport."""
 
 
69
        path = urlsplit(abspath)[2].lstrip('/')
 
 
70
        # Perform an XMLRPC request to resolve the path
 
 
71
        resolve = _request_factory(path)
 
 
72
        service = LaunchpadService()
 
 
74
            result = resolve.submit(service)
 
 
75
        except xmlrpclib.Fault, fault:
 
 
76
            raise errors.InvalidURL(
 
 
77
                path=abspath, extra=fault.faultString)
 
 
79
        if 'launchpad' in debug.debug_flags:
 
 
80
            trace.mutter("resolve_lp_path(%r) == %r", path, result)
 
 
83
            _lp_login = get_lp_login()
 
 
84
        for url in result['urls']:
 
 
85
            scheme, netloc, path, query, fragment = urlsplit(url)
 
 
86
            if scheme == 'bzr+ssh' and (netloc.endswith('launchpad.net') or
 
 
87
                                        netloc.endswith('launchpad.dev')):
 
 
88
                # Only accept launchpad.net bzr+ssh URLs if we know
 
 
89
                # the user's Launchpad login:
 
 
92
                url = urlunsplit((scheme, '%s@%s' % (_lp_login, netloc),
 
 
93
                                  path, query, fragment))
 
 
96
                # Use the URL if we can create a transport for it.
 
 
99
                except (errors.PathError, errors.TransportError):
 
 
104
            raise errors.InvalidURL(path=abspath,
 
 
105
                                    extra='no supported schemes')
 
 
108
    def _request_redirect(self, relpath):
 
 
109
        source = urlutils.join(self.base, relpath)
 
 
110
        # Split the source location into the branch location, and the
 
 
111
        # extra path components.
 
 
112
        pos = source.find('/.bzr/')
 
 
114
            branchpath = source[:pos]
 
 
119
        target = self._resolve(branchpath) + extra
 
 
120
        raise errors.RedirectRequested(
 
 
124
    def get(self, relpath):
 
 
125
        """See Transport.get()."""
 
 
126
        self._request_redirect(relpath)
 
 
128
    def mkdir(self, relpath, mode=None):
 
 
129
        """See Transport.mkdir()."""
 
 
130
        self._request_redirect(relpath)
 
 
133
def get_test_permutations():
 
 
134
    # Since this transport doesn't do anything once opened, it's not subjected
 
 
135
    # to the usual transport tests.