~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

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

  • Committer: John Arbash Meinel
  • Date: 2007-11-16 02:11:34 UTC
  • mto: This revision was merged to the branch mainline in revision 3002.
  • Revision ID: john@arbash-meinel.com-20071116021134-e0fjjpjllimkhj3v
Re-introduce the None check in case someone asks to uncommit *to* the last revision

Show diffs side-by-side

added added

removed removed

Lines of Context:
22
22
again.
23
23
"""
24
24
 
25
 
from urlparse import urlsplit, urlunsplit
26
 
import xmlrpclib
27
 
 
28
25
from bzrlib import (
29
 
    debug,
30
26
    errors,
31
 
    trace,
32
 
    urlutils,
33
27
    )
34
28
from bzrlib.transport import (
35
29
    get_transport,
36
 
    register_urlparse_netloc_protocol,
37
30
    Transport,
38
31
    )
39
32
 
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 _resolve(self, abspath,
66
 
                 _request_factory=ResolveLaunchpadPathRequest,
67
 
                 _lp_login=None):
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()
73
 
        try:
74
 
            result = resolve.submit(service)
75
 
        except xmlrpclib.Fault, fault:
76
 
            raise errors.InvalidURL(
77
 
                path=abspath, extra=fault.faultString)
78
 
 
79
 
        if 'launchpad' in debug.debug_flags:
80
 
            trace.mutter("resolve_lp_path(%r) == %r", path, result)
81
 
 
82
 
        if _lp_login is None:
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:
90
 
                if _lp_login is None:
91
 
                    continue
92
 
                url = urlunsplit((scheme, '%s@%s' % (_lp_login, netloc),
93
 
                                  path, query, fragment))
94
 
                break
95
 
            else:
96
 
                # Use the URL if we can create a transport for it.
97
 
                try:
98
 
                    get_transport(url)
99
 
                except (errors.PathError, errors.TransportError):
100
 
                    pass
101
 
                else:
102
 
                    break
103
 
        else:
104
 
            raise errors.InvalidURL(path=abspath,
105
 
                                    extra='no supported schemes')
106
 
        return url
107
 
 
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/')
113
 
        if pos >= 0:
114
 
            branchpath = source[:pos]
115
 
            extra = source[pos:]
116
 
        else:
117
 
            branchpath = source
118
 
            extra = ''
119
 
        target = self._resolve(branchpath) + extra
120
 
        raise errors.RedirectRequested(
121
 
            source=source,
122
 
            target=target)
123
 
 
124
 
    def get(self, relpath):
125
 
        """See Transport.get()."""
126
 
        self._request_redirect(relpath)
127
 
 
128
 
    def mkdir(self, relpath, mode=None):
129
 
        """See Transport.mkdir()."""
130
 
        self._request_redirect(relpath)
 
33
 
 
34
def launchpad_transport_indirect(base_url):
 
35
    """Uses Launchpad.net as a directory of open source software"""
 
36
    if base_url.startswith('lp:///'):
 
37
        real_url = 'http://code.launchpad.net/' + base_url[6:]
 
38
    elif base_url.startswith('lp:') and base_url[3] != '/':
 
39
        real_url = 'http://code.launchpad.net/' + base_url[3:]
 
40
    else:
 
41
        raise errors.InvalidURL(path=base_url)
 
42
    return get_transport(real_url)
131
43
 
132
44
 
133
45
def get_test_permutations():