← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

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

Merge register-branch plugin into default plugins directory

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/plugins/launchpad/lp_registration.py
 
1
# Copyright (C) 2006 by Canonical Ltd
 
2
#
 
3
# This program is free software; you can redistribute it and/or modify
 
4
# it under the terms of the GNU General Public License as published by
 
5
# the Free Software Foundation; either version 2 of the License, or
 
6
# (at your option) any later version.
 
7
#
 
8
# This program is distributed in the hope that it will be useful,
 
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the # GNU General Public License for more details.
 
11
#
 
12
# You should have received a copy of the GNU General Public License
 
13
# along with this program; if not, write to the Free Software
 
14
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
15
 
 
16
 
 
17
from getpass import getpass
 
18
import os
 
19
from urlparse import urlsplit, urlunsplit
 
20
import urllib
 
21
import xmlrpclib
 
22
 
 
23
import bzrlib.config
 
24
 
 
25
class LaunchpadService(object):
 
26
    """A service to talk to Launchpad via XMLRPC.
 
27
    
 
28
    See http://bazaar-vcs.org/Specs/LaunchpadRpc for the methods we can call.
 
29
    """
 
30
 
 
31
    # NB: this should always end in a slash to avoid xmlrpclib appending
 
32
    # '/RPC2'
 
33
    DEFAULT_SERVICE_URL = 'http://xmlrpc.launchpad.net/bazaar/'
 
34
 
 
35
    transport = None
 
36
    registrant_email = None
 
37
    registrant_password = None
 
38
 
 
39
 
 
40
    def __init__(self, transport=None):
 
41
        """Construct a new service talking to the launchpad rpc server"""
 
42
        if transport is None:
 
43
            uri_type = urllib.splittype(self.service_url)[0]
 
44
            if uri_type == 'https':
 
45
                transport = xmlrpclib.SafeTransport()
 
46
            else:
 
47
                transport = xmlrpclib.Transport()
 
48
            transport.user_agent = 'bzr/%s (xmlrpclib/%s)' \
 
49
                    % (bzrlib.__version__, xmlrpclib.__version__)
 
50
        self.transport = transport
 
51
 
 
52
 
 
53
    @property
 
54
    def service_url(self):
 
55
        """Return the http or https url for the xmlrpc server.
 
56
 
 
57
        This does not include the username/password credentials.
 
58
        """
 
59
        key = 'BZR_LP_XMLRPC_URL'
 
60
        if key in os.environ:
 
61
            return os.environ[key]
 
62
        else:
 
63
            return self.DEFAULT_SERVICE_URL
 
64
 
 
65
    def get_proxy(self):
 
66
        """Return the proxy for XMLRPC requests."""
 
67
        # auth info must be in url
 
68
        # TODO: if there's no registrant email perhaps we should just connect
 
69
        # anonymously?
 
70
        scheme, hostinfo, path = urlsplit(self.service_url)[:3]
 
71
        assert '@' not in hostinfo
 
72
        assert self.registrant_email is not None
 
73
        assert self.registrant_password is not None
 
74
        hostinfo = '%s:%s@%s' % (urllib.quote(self.registrant_email),
 
75
                                 urllib.quote(self.registrant_password),
 
76
                                 hostinfo)
 
77
        url = urlunsplit((scheme, hostinfo, path, '', ''))
 
78
        return xmlrpclib.ServerProxy(url, transport=self.transport)
 
79
 
 
80
    def gather_user_credentials(self):
 
81
        """Get the password from the user."""
 
82
        config = bzrlib.config.GlobalConfig()
 
83
        self.registrant_email = config.user_email()
 
84
        if self.registrant_password is None:
 
85
            prompt = 'launchpad.net password for %s: ' % \
 
86
                    self.registrant_email
 
87
            self.registrant_password = getpass(prompt)
 
88
 
 
89
    def send_request(self, method_name, method_params):
 
90
        proxy = self.get_proxy()
 
91
        assert method_name
 
92
        method = getattr(proxy, method_name)
 
93
        result = method(*method_params)
 
94
        return result
 
95
 
 
96
 
 
97
class BaseRequest(object):
 
98
    """Base request for talking to a XMLRPC server."""
 
99
 
 
100
    # Set this to the XMLRPC method name.
 
101
    _methodname = None
 
102
 
 
103
    def _request_params(self):
 
104
        """Return the arguments to pass to the method"""
 
105
        raise NotImplementedError(self._request_params)
 
106
 
 
107
    def submit(self, service):
 
108
        """Submit request to Launchpad XMLRPC server.
 
109
 
 
110
        :param service: LaunchpadService indicating where to send
 
111
            the request and the authentication credentials.
 
112
        """
 
113
        return service.send_request(self._methodname, self._request_params())
 
114
 
 
115
 
 
116
class BranchRegistrationRequest(BaseRequest):
 
117
    """Request to tell Launchpad about a bzr branch."""
 
118
 
 
119
    _methodname = 'register_branch'
 
120
 
 
121
    def __init__(self, branch_url,
 
122
                 branch_name='',
 
123
                 branch_title='',
 
124
                 branch_description='',
 
125
                 author_email='',
 
126
                 product_name='',
 
127
                 ):
 
128
        assert branch_url
 
129
        self.branch_url = branch_url
 
130
        if branch_name:
 
131
            self.branch_name = branch_name
 
132
        else:
 
133
            self.branch_name = self._find_default_branch_name(self.branch_url)
 
134
        self.branch_title = branch_title
 
135
        self.branch_description = branch_description
 
136
        self.author_email = author_email
 
137
        self.product_name = product_name
 
138
 
 
139
    def _request_params(self):
 
140
        """Return xmlrpc request parameters"""
 
141
        # This must match the parameter tuple expected by Launchpad for this
 
142
        # method
 
143
        return (self.branch_url,
 
144
                self.branch_name,
 
145
                self.branch_title,
 
146
                self.branch_description,
 
147
                self.author_email,
 
148
                self.product_name,
 
149
               )
 
150
 
 
151
    def _find_default_branch_name(self, branch_url):
 
152
        i = branch_url.rfind('/')
 
153
        return branch_url[i+1:]
 
154
 
 
155
 
 
156
class BranchBugLinkRequest(BaseRequest):
 
157
    """Request to link a bzr branch in Launchpad to a bug."""
 
158
 
 
159
    _methodname = 'link_branch_to_bug'
 
160
 
 
161
    def __init__(self, branch_url, bug_id):
 
162
        assert branch_url
 
163
        self.bug_id = bug_id
 
164
        self.branch_url = branch_url
 
165
 
 
166
    def _request_params(self):
 
167
        """Return xmlrpc request parameters"""
 
168
        # This must match the parameter tuple expected by Launchpad for this
 
169
        # method
 
170
        return (self.branch_url, self.bug_id, '')