1
# Copyright (C) 2007 Canonical Ltd
1
# Copyright (C) 2007, 2008 Canonical Ltd
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
15
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
18
"""Directory lookup that uses Launchpad."""
25
20
from urlparse import urlsplit, urlunsplit
48
42
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
self.lp_instance = urlsplit(base)[1]
62
if self.lp_instance == '':
63
self.lp_instance = None
64
elif self.lp_instance not in LaunchpadService.LAUNCHPAD_INSTANCE:
65
raise errors.InvalidURL(path=base)
45
class LaunchpadDirectory(object):
67
47
def _requires_launchpad_login(self, scheme, netloc, path, query,
75
55
and (netloc.endswith('launchpad.net')
76
56
or netloc.endswith('launchpad.dev')))
78
def _resolve(self, abspath,
58
def look_up(self, name, url):
59
"""See DirectoryService.look_up"""
60
return self._resolve(url)
62
def _resolve(self, url,
79
63
_request_factory=ResolveLaunchpadPathRequest,
81
65
"""Resolve the base URL for this transport."""
82
path = urlsplit(abspath)[2].lstrip('/')
66
result = urlsplit(url)
83
67
# Perform an XMLRPC request to resolve the path
84
resolve = _request_factory(path)
85
service = LaunchpadService(lp_instance=self.lp_instance)
68
lp_instance = result[1]
71
elif lp_instance not in LaunchpadService.LAUNCHPAD_INSTANCE:
72
raise errors.InvalidURL(path=url)
73
resolve = _request_factory(result[2].strip('/'))
74
service = LaunchpadService(lp_instance=lp_instance)
87
76
result = resolve.submit(service)
88
77
except xmlrpclib.Fault, fault:
89
78
raise errors.InvalidURL(
90
path=abspath, extra=fault.faultString)
79
path=url, extra=fault.faultString)
92
81
if 'launchpad' in debug.debug_flags:
93
82
trace.mutter("resolve_lp_path(%r) == %r", path, result)
95
84
if _lp_login is None:
96
85
_lp_login = get_lp_login()
97
87
for url in result['urls']:
98
88
scheme, netloc, path, query, fragment = urlsplit(url)
99
89
if self._requires_launchpad_login(scheme, netloc, path, query,
101
91
# Only accept launchpad.net bzr+ssh URLs if we know
102
92
# the user's Launchpad login:
103
93
if _lp_login is None:
95
trace.warning('You have not informed bzr of your '
96
'launchpad login. If you are attempting a '
97
'write operation it may fail. If it does, '
98
'run "bzr launchpad-login YOUR_ID" to '
99
'set your login and try again.')
105
102
url = urlunsplit((scheme, '%s@%s' % (_lp_login, netloc),
106
103
path, query, fragment))
117
raise errors.InvalidURL(path=abspath,
118
extra='no supported schemes')
114
raise errors.InvalidURL(path=url, extra='no supported schemes')
121
def _request_redirect(self, relpath):
122
source = urlutils.join(self.base, relpath)
123
# Split the source location into the branch location, and the
124
# extra path components.
125
pos = source.find('/.bzr/')
127
branchpath = source[:pos]
132
target = self._resolve(branchpath) + extra
133
raise errors.RedirectRequested(
137
def get(self, relpath):
138
"""See Transport.get()."""
139
self._request_redirect(relpath)
141
def mkdir(self, relpath, mode=None):
142
"""See Transport.mkdir()."""
143
self._request_redirect(relpath)
146
118
def get_test_permutations():
147
119
# Since this transport doesn't do anything once opened, it's not subjected
added
removed
bzrlib/plugins/launchpad/lp_directory.py
