1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
|
# Copyright (C) 2007, 2008 Canonical Ltd
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
"""Directory lookup that uses Launchpad."""
from urlparse import urlsplit, urlunsplit
import xmlrpclib
from bzrlib import (
debug,
errors,
trace,
urlutils,
)
from bzrlib.transport import (
get_transport,
register_urlparse_netloc_protocol,
)
from bzrlib.plugins.launchpad.lp_registration import (
LaunchpadService, ResolveLaunchpadPathRequest)
from bzrlib.plugins.launchpad.account import get_lp_login
# As bzrlib.transport.remote may not be loaded yet, make sure bzr+ssh
# is counted as a netloc protocol.
register_urlparse_netloc_protocol('bzr+ssh')
register_urlparse_netloc_protocol('lp')
class LaunchpadDirectory(object):
def _requires_launchpad_login(self, scheme, netloc, path, query,
fragment):
"""Does the URL require a Launchpad login in order to be reached?
The URL is specified by its parsed components, as returned from
urlsplit.
"""
return (scheme in ('bzr+ssh', 'sftp')
and (netloc.endswith('launchpad.net')
or netloc.endswith('launchpad.dev')))
def look_up(self, name, url):
"""See DirectoryService.look_up"""
return self._resolve(url)
def _resolve(self, url,
_request_factory=ResolveLaunchpadPathRequest,
_lp_login=None):
"""Resolve the base URL for this transport."""
result = urlsplit(url)
# Perform an XMLRPC request to resolve the path
lp_instance = result[1]
if lp_instance == '':
lp_instance = None
elif lp_instance not in LaunchpadService.LAUNCHPAD_INSTANCE:
raise errors.InvalidURL(path=url)
resolve = _request_factory(result[2].strip('/'))
service = LaunchpadService(lp_instance=lp_instance)
try:
result = resolve.submit(service)
except xmlrpclib.Fault, fault:
raise errors.InvalidURL(
path=url, extra=fault.faultString)
if 'launchpad' in debug.debug_flags:
trace.mutter("resolve_lp_path(%r) == %r", path, result)
if _lp_login is None:
_lp_login = get_lp_login()
_warned_login = False
for url in result['urls']:
scheme, netloc, path, query, fragment = urlsplit(url)
if self._requires_launchpad_login(scheme, netloc, path, query,
fragment):
# Only accept launchpad.net bzr+ssh URLs if we know
# the user's Launchpad login:
if _lp_login is None:
if not _warned_login:
trace.warning('You have not informed bzr of your '
'launchpad login. If you are attempting a\n'
'write operation and it fails, run '
'"bzr launchpad-login YOUR_ID" and try again.')
_warned_login = True
continue
url = urlunsplit((scheme, '%s@%s' % (_lp_login, netloc),
path, query, fragment))
break
else:
# Use the URL if we can create a transport for it.
try:
get_transport(url)
except (errors.PathError, errors.TransportError):
pass
else:
break
else:
raise errors.InvalidURL(path=url, extra='no supported schemes')
return url
def get_test_permutations():
# Since this transport doesn't do anything once opened, it's not subjected
# to the usual transport tests.
return []
|