← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

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

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2007-12-07 03:30:29 UTC
  • mfrom: (3079.1.1 criss-cross)
  • Revision ID: pqm@pqm.ubuntu.com-20071207033029-7tx9ezbg3nlk3io1
(Alexander Belchenko) topic for criss-cross should have title,
        otherwise autogenerated bzr_man has problems

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/plugins/launchpad/lp_indirect.py
 
1
# Copyright (C) 2007 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
 
 
17
 
 
18
"""Transport indirection that uses Launchpad as a directory lookup.
 
19
 
 
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
 
22
again.
 
23
"""
 
24
 
 
25
from urlparse import urlsplit, urlunsplit
 
26
import xmlrpclib
 
27
 
 
28
from bzrlib import (
 
29
    debug,
 
30
    errors,
 
31
    trace,
 
32
    urlutils,
 
33
    )
 
34
from bzrlib.transport import (
 
35
    get_transport,
 
36
    register_urlparse_netloc_protocol,
 
37
    Transport,
 
38
    )
 
39
 
 
40
from bzrlib.plugins.launchpad.lp_registration import (
 
41
    LaunchpadService, ResolveLaunchpadPathRequest)
 
42
from bzrlib.plugins.launchpad.account import get_lp_login
 
43
 
 
44
 
 
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')
 
49
 
 
50
 
 
51
class LaunchpadTransport(Transport):
 
52
    """lp:/// URL transport
 
53
 
 
54
    This transport redirects requests to the real branch location
 
55
    after resolving the URL via an XMLRPC request to Launchpad.
 
56
    """
 
57
 
 
58
    def __init__(self, base):
 
59
        super(LaunchpadTransport, self).__init__(base)
 
60
        # We only support URLs without a netloc
 
61
        netloc = urlsplit(base)[1]
 
62
        if netloc != '':
 
63
            raise errors.InvalidURL(path=base)
 
64
 
 
65
    def _requires_launchpad_login(self, scheme, netloc, path, query,
 
66
                                  fragment):
 
67
        """Does the URL require a Launchpad login in order to be reached?
 
68
 
 
69
        The URL is specified by its parsed components, as returned from
 
70
        urlsplit.
 
71
        """
 
72
        return (scheme in ('bzr+ssh', 'sftp')
 
73
                and (netloc.endswith('launchpad.net')
 
74
                     or netloc.endswith('launchpad.dev')))
 
75
 
 
76
    def _resolve(self, abspath,
 
77
                 _request_factory=ResolveLaunchpadPathRequest,
 
78
                 _lp_login=None):
 
79
        """Resolve the base URL for this transport."""
 
80
        path = urlsplit(abspath)[2].lstrip('/')
 
81
        # Perform an XMLRPC request to resolve the path
 
82
        resolve = _request_factory(path)
 
83
        service = LaunchpadService()
 
84
        try:
 
85
            result = resolve.submit(service)
 
86
        except xmlrpclib.Fault, fault:
 
87
            raise errors.InvalidURL(
 
88
                path=abspath, extra=fault.faultString)
 
89
 
 
90
        if 'launchpad' in debug.debug_flags:
 
91
            trace.mutter("resolve_lp_path(%r) == %r", path, result)
 
92
 
 
93
        if _lp_login is None:
 
94
            _lp_login = get_lp_login()
 
95
        for url in result['urls']:
 
96
            scheme, netloc, path, query, fragment = urlsplit(url)
 
97
            if self._requires_launchpad_login(scheme, netloc, path, query,
 
98
                                              fragment):
 
99
                # Only accept launchpad.net bzr+ssh URLs if we know
 
100
                # the user's Launchpad login:
 
101
                if _lp_login is None:
 
102
                    continue
 
103
                url = urlunsplit((scheme, '%s@%s' % (_lp_login, netloc),
 
104
                                  path, query, fragment))
 
105
                break
 
106
            else:
 
107
                # Use the URL if we can create a transport for it.
 
108
                try:
 
109
                    get_transport(url)
 
110
                except (errors.PathError, errors.TransportError):
 
111
                    pass
 
112
                else:
 
113
                    break
 
114
        else:
 
115
            raise errors.InvalidURL(path=abspath,
 
116
                                    extra='no supported schemes')
 
117
        return url
 
118
 
 
119
    def _request_redirect(self, relpath):
 
120
        source = urlutils.join(self.base, relpath)
 
121
        # Split the source location into the branch location, and the
 
122
        # extra path components.
 
123
        pos = source.find('/.bzr/')
 
124
        if pos >= 0:
 
125
            branchpath = source[:pos]
 
126
            extra = source[pos:]
 
127
        else:
 
128
            branchpath = source
 
129
            extra = ''
 
130
        target = self._resolve(branchpath) + extra
 
131
        raise errors.RedirectRequested(
 
132
            source=source,
 
133
            target=target)
 
134
 
 
135
    def get(self, relpath):
 
136
        """See Transport.get()."""
 
137
        self._request_redirect(relpath)
 
138
 
 
139
    def mkdir(self, relpath, mode=None):
 
140
        """See Transport.mkdir()."""
 
141
        self._request_redirect(relpath)
 
142
 
 
143
 
 
144
def get_test_permutations():
 
145
    # Since this transport doesn't do anything once opened, it's not subjected
 
146
    # to the usual transport tests.
 
147
    return []