13
13
# You should have received a copy of the GNU General Public License
14
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
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17
18
"""Directory lookup that uses Launchpad."""
19
from __future__ import absolute_import
21
from urlparse import urlsplit
20
from urlparse import urlsplit, urlunsplit
24
23
from bzrlib import (
30
from bzrlib.i18n import gettext
29
from bzrlib.transport import (
31
register_urlparse_netloc_protocol,
32
34
from bzrlib.plugins.launchpad.lp_registration import (
33
35
LaunchpadService, ResolveLaunchpadPathRequest)
67
59
"""See DirectoryService.look_up"""
68
60
return self._resolve(url)
70
def _resolve_locally(self, path, url, _request_factory):
71
# This is the best I could work out about XMLRPC. If an lp: url
72
# includes ~user, then it is specially validated. Otherwise, it is just
73
# sent to +branch/$path.
74
_, netloc, _, _, _ = urlsplit(url)
76
netloc = LaunchpadService.DEFAULT_INSTANCE
77
base_url = LaunchpadService.LAUNCHPAD_DOMAINS[netloc]
78
base = 'bzr+ssh://bazaar.%s/' % (base_url,)
80
if path.startswith('~'):
81
# A ~user style path, validate it a bit.
82
# If a path looks fishy, fall back to asking XMLRPC to
83
# resolve it for us. That way we still get their nicer error
85
parts = path.split('/')
87
or (parts[1] in ('ubuntu', 'debian') and len(parts) < 5)):
88
# This special case requires 5-parts to be valid.
93
return self._resolve_via_xmlrpc(path, url, _request_factory)
94
return {'urls': [base + path]}
96
def _resolve_via_xmlrpc(self, path, url, _request_factory):
97
service = LaunchpadService.for_url(url)
98
resolve = _request_factory(path)
62
def _resolve(self, url,
63
_request_factory=ResolveLaunchpadPathRequest,
65
"""Resolve the base URL for this transport."""
66
result = urlsplit(url)
67
# Perform an XMLRPC request to resolve the path
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)
100
76
result = resolve.submit(service)
101
77
except xmlrpclib.Fault, fault:
102
78
raise errors.InvalidURL(
103
79
path=url, extra=fault.faultString)
106
def _update_url_scheme(self, url):
107
# Do ubuntu: and debianlp: expansions.
108
scheme, netloc, path, query, fragment = urlsplit(url)
109
if scheme in ('ubuntu', 'debianlp'):
110
if scheme == 'ubuntu':
112
distro_series = _ubuntu_series_shortcuts
113
elif scheme == 'debianlp':
115
# No shortcuts for Debian distroseries.
118
raise AssertionError('scheme should be ubuntu: or debianlp:')
119
# Split the path. It's either going to be 'project' or
120
# 'series/project', but recognize that it may be a series we don't
122
path_parts = path.split('/')
123
if len(path_parts) == 1:
124
# It's just a project name.
125
lp_url_template = 'lp:%(distro)s/%(project)s'
126
project = path_parts[0]
128
elif len(path_parts) == 2:
129
# It's a series and project.
130
lp_url_template = 'lp:%(distro)s/%(series)s/%(project)s'
131
series, project = path_parts
133
# There are either 0 or > 2 path parts, neither of which is
134
# supported for these schemes.
135
raise errors.InvalidURL('Bad path: %s' % url)
136
# Expand any series shortcuts, but keep unknown series.
137
series = distro_series.get(series, series)
138
# Hack the url and let the following do the final resolution.
139
url = lp_url_template % dict(
143
scheme, netloc, path, query, fragment = urlsplit(url)
146
def _expand_user(self, path, url, lp_login):
147
if path.startswith('~/'):
149
raise errors.InvalidURL(path=url,
150
extra='Cannot resolve "~" to your username.'
151
' See "bzr help launchpad-login"')
152
path = '~' + lp_login + path[1:]
155
def _resolve(self, url,
156
_request_factory=ResolveLaunchpadPathRequest,
158
"""Resolve the base URL for this transport."""
159
url, path = self._update_url_scheme(url)
81
if 'launchpad' in debug.debug_flags:
82
trace.mutter("resolve_lp_path(%r) == %r", path, result)
160
84
if _lp_login is None:
161
85
_lp_login = get_lp_login()
162
path = path.strip('/')
163
path = self._expand_user(path, url, _lp_login)
164
if _lp_login is not None:
165
result = self._resolve_locally(path, url, _request_factory)
166
if 'launchpad' in debug.debug_flags:
168
result = self._resolve_via_xmlrpc(path, url, _request_factory)
170
'resolution for {0}\n local: {1}\n remote: {2}').format(
171
url, local_res['urls'], result['urls']))
173
result = self._resolve_via_xmlrpc(path, url, _request_factory)
175
if 'launchpad' in debug.debug_flags:
176
trace.mutter("resolve_lp_path(%r) == %r", url, result)
178
86
_warned_login = False
179
87
for url in result['urls']:
180
88
scheme, netloc, path, query, fragment = urlsplit(url)