1
# Copyright (C) 2006-2010 Canonical Ltd
1
# Copyright (C) 2006 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
24
from bzrlib.lazy_import import lazy_import
25
lazy_import(globals(), """
26
from bzrlib import urlutils
24
29
from bzrlib import (
28
32
__version__ as _bzrlib_version,
30
from bzrlib.transport.http import _urllib2_wrappers
51
53
errors.BzrError.__init__(self, url=url)
54
class XMLRPCTransport(xmlrpclib.Transport):
56
def __init__(self, scheme):
57
# In python2.4 xmlrpclib.Transport is a old-style class, and does not
58
# define __init__, so we check first
59
init = getattr(xmlrpclib.Transport, '__init__', None)
63
self._opener = _urllib2_wrappers.Opener()
66
def request(self, host, handler, request_body, verbose=0):
67
self.verbose = verbose
68
url = self._scheme + "://" + host + handler
69
request = _urllib2_wrappers.Request("POST", url, request_body)
70
# FIXME: _urllib2_wrappers will override user-agent with its own
71
# request.add_header("User-Agent", self.user_agent)
72
request.add_header("Content-Type", "text/xml")
74
response = self._opener.open(request)
75
if response.code != 200:
76
raise xmlrpclib.ProtocolError(host + handler, response.code,
77
response.msg, response.info())
78
return self.parse_response(response)
81
56
class LaunchpadService(object):
82
57
"""A service to talk to Launchpad via XMLRPC.
84
See http://wiki.bazaar.canonical.com/Specs/LaunchpadRpc for the methods we can call.
59
See http://bazaar-vcs.org/Specs/LaunchpadRpc for the methods we can call.
87
62
LAUNCHPAD_DOMAINS = {
115
90
self._lp_instance = lp_instance
116
91
if transport is None:
117
92
uri_type = urllib.splittype(self.service_url)[0]
118
transport = XMLRPCTransport(uri_type)
93
if uri_type == 'https':
94
transport = xmlrpclib.SafeTransport()
96
transport = xmlrpclib.Transport()
119
97
transport.user_agent = 'bzr/%s (xmlrpclib/%s)' \
120
98
% (_bzrlib_version, xmlrpclib.__version__)
121
99
self.transport = transport
124
103
def service_url(self):
125
104
"""Return the http or https url for the xmlrpc server.
138
117
return self.DEFAULT_SERVICE_URL
141
def for_url(cls, url, **kwargs):
142
"""Return the Launchpad service corresponding to the given URL."""
143
result = urlsplit(url)
144
lp_instance = result[1]
145
if lp_instance == '':
147
elif lp_instance not in cls.LAUNCHPAD_INSTANCE:
148
raise errors.InvalidURL(path=url)
149
return cls(lp_instance=lp_instance, **kwargs)
151
119
def get_proxy(self, authenticated):
152
120
"""Return the proxy for XMLRPC requests."""
153
121
if authenticated:
223
191
instance = self._lp_instance
224
192
return self.LAUNCHPAD_DOMAINS[instance]
226
def _guess_branch_path(self, branch_url, _request_factory=None):
194
def get_web_url_from_branch_url(self, branch_url, _request_factory=None):
195
"""Get the Launchpad web URL for the given branch URL.
197
:raise errors.InvalidURL: if 'branch_url' cannot be identified as a
198
Launchpad branch URL.
199
:return: The URL of the branch on Launchpad.
227
201
scheme, hostinfo, path = urlsplit(branch_url)[:3]
228
202
if _request_factory is None:
229
203
_request_factory = ResolveLaunchpadPathRequest
241
215
for domain in self.LAUNCHPAD_DOMAINS.itervalues())
242
216
if hostinfo not in domains:
243
217
raise NotLaunchpadBranch(branch_url)
244
return path.lstrip('/')
246
def get_web_url_from_branch_url(self, branch_url, _request_factory=None):
247
"""Get the Launchpad web URL for the given branch URL.
249
:raise errors.InvalidURL: if 'branch_url' cannot be identified as a
250
Launchpad branch URL.
251
:return: The URL of the branch on Launchpad.
253
path = self._guess_branch_path(branch_url, _request_factory)
254
218
return urlutils.join('https://code.%s' % self.domain, path)