1
# Copyright (C) 2006 Canonical Ltd
1
# Copyright (C) 2006-2010 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
29
24
from bzrlib import (
32
28
__version__ as _bzrlib_version,
30
from bzrlib.transport.http import _urllib2_wrappers
53
51
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)
56
81
class LaunchpadService(object):
57
82
"""A service to talk to Launchpad via XMLRPC.
90
115
self._lp_instance = lp_instance
91
116
if transport is None:
92
117
uri_type = urllib.splittype(self.service_url)[0]
93
if uri_type == 'https':
94
transport = xmlrpclib.SafeTransport()
96
transport = xmlrpclib.Transport()
118
transport = XMLRPCTransport(uri_type)
97
119
transport.user_agent = 'bzr/%s (xmlrpclib/%s)' \
98
120
% (_bzrlib_version, xmlrpclib.__version__)
99
121
self.transport = transport
103
124
def service_url(self):
104
125
"""Return the http or https url for the xmlrpc server.
117
138
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)
119
151
def get_proxy(self, authenticated):
120
152
"""Return the proxy for XMLRPC requests."""
121
153
if authenticated:
191
223
instance = self._lp_instance
192
224
return self.LAUNCHPAD_DOMAINS[instance]
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.
226
def _guess_branch_path(self, branch_url, _request_factory=None):
201
227
scheme, hostinfo, path = urlsplit(branch_url)[:3]
202
228
if _request_factory is None:
203
229
_request_factory = ResolveLaunchpadPathRequest
215
241
for domain in self.LAUNCHPAD_DOMAINS.itervalues())
216
242
if hostinfo not in domains:
217
243
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)
218
254
return urlutils.join('https://code.%s' % self.domain, path)