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
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
|
# Copyright (C) 2007 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
"""Transport indirection that uses Launchpad as a directory lookup.
When the transport is opened, it immediately redirects to a url
on Launchpad, which can then either serve the branch itself or redirect
again.
"""
from urlparse import urlsplit, urlunsplit
import xmlrpclib
from bzrlib import (
debug,
errors,
trace,
urlutils,
)
from bzrlib.transport import (
get_transport,
register_urlparse_netloc_protocol,
Transport,
)
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 LaunchpadTransport(Transport):
"""lp:/// URL transport
This transport redirects requests to the real branch location
after resolving the URL via an XMLRPC request to Launchpad.
"""
def __init__(self, base):
super(LaunchpadTransport, self).__init__(base)
# We only support URLs without a netloc
netloc = urlsplit(base)[1]
if netloc != '':
raise errors.InvalidURL(path=base)
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 _resolve(self, abspath,
_request_factory=ResolveLaunchpadPathRequest,
_lp_login=None):
"""Resolve the base URL for this transport."""
path = urlsplit(abspath)[2].lstrip('/')
# Perform an XMLRPC request to resolve the path
resolve = _request_factory(path)
service = LaunchpadService()
try:
result = resolve.submit(service)
except xmlrpclib.Fault, fault:
raise errors.InvalidURL(
path=abspath, 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()
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:
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=abspath,
extra='no supported schemes')
return url
def _request_redirect(self, relpath):
source = urlutils.join(self.base, relpath)
# Split the source location into the branch location, and the
# extra path components.
pos = source.find('/.bzr/')
if pos >= 0:
branchpath = source[:pos]
extra = source[pos:]
else:
branchpath = source
extra = ''
target = self._resolve(branchpath) + extra
raise errors.RedirectRequested(
source=source,
target=target)
def get(self, relpath):
"""See Transport.get()."""
self._request_redirect(relpath)
def mkdir(self, relpath, mode=None):
"""See Transport.mkdir()."""
self._request_redirect(relpath)
def get_test_permutations():
# Since this transport doesn't do anything once opened, it's not subjected
# to the usual transport tests.
return []
|