/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/plugins/launchpad/lp_registration.py

  • Committer: James Henstridge
  • Date: 2007-10-31 19:47:45 UTC
  • mto: This revision was merged to the branch mainline in revision 2980.
  • Revision ID: james@jamesh.id.au-20071031194745-u5nlc14tr74w7t3g
Use get_transport() to decide whether Bazaar supports a given URL.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/plugins/launchpad/lp_registration.py
1
 
# Copyright (C) 2006-2010 Canonical Ltd
 
1
# Copyright (C) 2006 Canonical Ltd
2
2
#
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
12
12
#
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
16
 
 
17
 
 
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
16
 
 
17
 
 
18
from getpass import getpass
18
19
import os
19
 
import socket
20
20
from urlparse import urlsplit, urlunsplit
21
21
import urllib
22
22
import xmlrpclib
23
23
 
24
 
from bzrlib import (
25
 
    config,
26
 
    errors,
27
 
    urlutils,
28
 
    __version__ as _bzrlib_version,
29
 
    )
30
 
from bzrlib.transport.http import _urllib2_wrappers
31
 
 
 
24
import bzrlib.config
 
25
import bzrlib.errors as errors
32
26
 
33
27
# for testing, do
34
28
'''
35
29
export BZR_LP_XMLRPC_URL=http://xmlrpc.staging.launchpad.net/bazaar/
36
30
'''
37
31
 
38
 
class InvalidLaunchpadInstance(errors.BzrError):
39
 
 
40
 
    _fmt = "%(lp_instance)s is not a valid Launchpad instance."
41
 
 
42
 
    def __init__(self, lp_instance):
43
 
        errors.BzrError.__init__(self, lp_instance=lp_instance)
44
 
 
45
 
 
46
 
class NotLaunchpadBranch(errors.BzrError):
47
 
 
48
 
    _fmt = "%(url)s is not registered on Launchpad."
49
 
 
50
 
    def __init__(self, url):
51
 
        errors.BzrError.__init__(self, url=url)
52
 
 
53
 
 
54
 
class XMLRPCTransport(xmlrpclib.Transport):
55
 
 
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)
60
 
        if init is not None:
61
 
            init(self)
62
 
        self._scheme = scheme
63
 
        self._opener = _urllib2_wrappers.Opener()
64
 
        self.verbose = 0
65
 
 
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")
73
 
 
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)
79
 
 
80
 
 
81
32
class LaunchpadService(object):
82
33
    """A service to talk to Launchpad via XMLRPC.
83
 
 
84
 
    See http://wiki.bazaar.canonical.com/Specs/LaunchpadRpc for the methods we can call.
 
34
    
 
35
    See http://bazaar-vcs.org/Specs/LaunchpadRpc for the methods we can call.
85
36
    """
86
37
 
87
 
    LAUNCHPAD_DOMAINS = {
88
 
        'production': 'launchpad.net',
89
 
        'edge': 'edge.launchpad.net',
90
 
        'staging': 'staging.launchpad.net',
91
 
        'demo': 'demo.launchpad.net',
92
 
        'dev': 'launchpad.dev',
93
 
        }
94
 
 
95
 
    # NB: these should always end in a slash to avoid xmlrpclib appending
 
38
    # NB: this should always end in a slash to avoid xmlrpclib appending
96
39
    # '/RPC2'
97
 
    LAUNCHPAD_INSTANCE = {}
98
 
    for instance, domain in LAUNCHPAD_DOMAINS.iteritems():
99
 
        LAUNCHPAD_INSTANCE[instance] = 'https://xmlrpc.%s/bazaar/' % domain
100
 
 
101
 
    # We use edge as the default because:
102
 
    # Beta users get redirected to it
103
 
    # All users can use it
104
 
    # There is a bug in the launchpad side where redirection causes an OOPS.
105
 
    DEFAULT_INSTANCE = 'edge'
106
 
    DEFAULT_SERVICE_URL = LAUNCHPAD_INSTANCE[DEFAULT_INSTANCE]
 
40
    DEFAULT_SERVICE_URL = 'https://xmlrpc.launchpad.net/bazaar/'
107
41
 
108
42
    transport = None
109
43
    registrant_email = None
110
44
    registrant_password = None
111
45
 
112
46
 
113
 
    def __init__(self, transport=None, lp_instance=None):
 
47
    def __init__(self, transport=None):
114
48
        """Construct a new service talking to the launchpad rpc server"""
115
 
        self._lp_instance = lp_instance
116
49
        if transport is None:
117
50
            uri_type = urllib.splittype(self.service_url)[0]
118
 
            transport = XMLRPCTransport(uri_type)
 
51
            if uri_type == 'https':
 
52
                transport = xmlrpclib.SafeTransport()
 
53
            else:
 
54
                transport = xmlrpclib.Transport()
119
55
            transport.user_agent = 'bzr/%s (xmlrpclib/%s)' \
120
 
                    % (_bzrlib_version, xmlrpclib.__version__)
 
56
                    % (bzrlib.__version__, xmlrpclib.__version__)
121
57
        self.transport = transport
122
58
 
 
59
 
123
60
    @property
124
61
    def service_url(self):
125
62
        """Return the http or https url for the xmlrpc server.
129
66
        key = 'BZR_LP_XMLRPC_URL'
130
67
        if key in os.environ:
131
68
            return os.environ[key]
132
 
        elif self._lp_instance is not None:
133
 
            try:
134
 
                return self.LAUNCHPAD_INSTANCE[self._lp_instance]
135
 
            except KeyError:
136
 
                raise InvalidLaunchpadInstance(self._lp_instance)
137
69
        else:
138
70
            return self.DEFAULT_SERVICE_URL
139
71
 
140
 
    @classmethod
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 == '':
146
 
            lp_instance = None
147
 
        elif lp_instance not in cls.LAUNCHPAD_INSTANCE:
148
 
            raise errors.InvalidURL(path=url)
149
 
        return cls(lp_instance=lp_instance, **kwargs)
150
 
 
151
72
    def get_proxy(self, authenticated):
152
73
        """Return the proxy for XMLRPC requests."""
153
74
        if authenticated:
155
76
            # TODO: if there's no registrant email perhaps we should
156
77
            # just connect anonymously?
157
78
            scheme, hostinfo, path = urlsplit(self.service_url)[:3]
158
 
            if '@' in hostinfo:
159
 
                raise AssertionError(hostinfo)
160
 
            if self.registrant_email is None:
161
 
                raise AssertionError()
162
 
            if self.registrant_password is None:
163
 
                raise AssertionError()
 
79
            assert '@' not in hostinfo
 
80
            assert self.registrant_email is not None
 
81
            assert self.registrant_password is not None
164
82
            # TODO: perhaps fully quote the password to make it very slightly
165
83
            # obscured
166
84
            # TODO: can we perhaps add extra Authorization headers
178
96
 
179
97
    def gather_user_credentials(self):
180
98
        """Get the password from the user."""
181
 
        the_config = config.GlobalConfig()
182
 
        self.registrant_email = the_config.user_email()
 
99
        config = bzrlib.config.GlobalConfig()
 
100
        self.registrant_email = config.user_email()
183
101
        if self.registrant_password is None:
184
 
            auth = config.AuthenticationConfig()
185
 
            scheme, hostinfo = urlsplit(self.service_url)[:2]
186
102
            prompt = 'launchpad.net password for %s: ' % \
187
103
                    self.registrant_email
188
 
            # We will reuse http[s] credentials if we can, prompt user
189
 
            # otherwise
190
 
            self.registrant_password = auth.get_password(scheme, hostinfo,
191
 
                                                         self.registrant_email,
192
 
                                                         prompt=prompt)
 
104
            self.registrant_password = getpass(prompt)
193
105
 
194
106
    def send_request(self, method_name, method_params, authenticated):
195
107
        proxy = self.get_proxy(authenticated)
 
108
        assert method_name
196
109
        method = getattr(proxy, method_name)
197
110
        try:
198
111
            result = method(*method_params)
209
122
                # TODO: print more headers to help in tracking down failures
210
123
                raise errors.BzrError("xmlrpc protocol error connecting to %s: %s %s"
211
124
                        % (self.service_url, e.errcode, e.errmsg))
212
 
        except socket.gaierror, e:
213
 
            raise errors.ConnectionError(
214
 
                "Could not resolve '%s'" % self.domain,
215
 
                orig_error=e)
216
125
        return result
217
126
 
218
 
    @property
219
 
    def domain(self):
220
 
        if self._lp_instance is None:
221
 
            instance = self.DEFAULT_INSTANCE
222
 
        else:
223
 
            instance = self._lp_instance
224
 
        return self.LAUNCHPAD_DOMAINS[instance]
225
 
 
226
 
    def _guess_branch_path(self, branch_url, _request_factory=None):
227
 
        scheme, hostinfo, path = urlsplit(branch_url)[:3]
228
 
        if _request_factory is None:
229
 
            _request_factory = ResolveLaunchpadPathRequest
230
 
        if scheme == 'lp':
231
 
            resolve = _request_factory(path)
232
 
            try:
233
 
                result = resolve.submit(self)
234
 
            except xmlrpclib.Fault, fault:
235
 
                raise errors.InvalidURL(branch_url, str(fault))
236
 
            branch_url = result['urls'][0]
237
 
            path = urlsplit(branch_url)[2]
238
 
        else:
239
 
            domains = (
240
 
                'bazaar.%s' % domain
241
 
                for domain in self.LAUNCHPAD_DOMAINS.itervalues())
242
 
            if hostinfo not in domains:
243
 
                raise NotLaunchpadBranch(branch_url)
244
 
        return path.lstrip('/')
245
 
 
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.
248
 
 
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.
252
 
        """
253
 
        path = self._guess_branch_path(branch_url, _request_factory)
254
 
        return urlutils.join('https://code.%s' % self.domain, path)
255
 
 
256
127
 
257
128
class BaseRequest(object):
258
129
    """Base request for talking to a XMLRPC server."""
277
148
 
278
149
class DryRunLaunchpadService(LaunchpadService):
279
150
    """Service that just absorbs requests without sending to server.
280
 
 
 
151
    
281
152
    The dummy service does not need authentication.
282
153
    """
283
154
 
300
171
                 author_email='',
301
172
                 product_name='',
302
173
                 ):
303
 
        if not branch_url:
304
 
            raise errors.InvalidURL(branch_url, "You need to specify a non-empty branch URL.")
 
174
        assert branch_url
305
175
        self.branch_url = branch_url
306
176
        if branch_name:
307
177
            self.branch_name = branch_name
335
205
    _methodname = 'link_branch_to_bug'
336
206
 
337
207
    def __init__(self, branch_url, bug_id):
 
208
        assert branch_url
338
209
        self.bug_id = bug_id
339
210
        self.branch_url = branch_url
340
211
 
352
223
    _authenticated = False
353
224
 
354
225
    def __init__(self, path):
355
 
        if not path:
356
 
            raise errors.InvalidURL(path=path,
357
 
                                    extra="You must specify a project.")
 
226
        assert path
358
227
        self.path = path
359
228
 
360
229
    def _request_params(self):