← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

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

  • Committer: John Arbash Meinel
  • Date: 2012-09-19 07:58:27 UTC
  • mfrom: (6437.63.9 2.5)
  • mto: This revision was merged to the branch mainline in revision 6563.
  • Revision ID: john@arbash-meinel.com-20120919075827-36b2b042kiaps0d3
Merge bzr-2.5.2 into trunk to get the fixes for ConnectionReset.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/plugins/launchpad/test_register.py
1
 
# Copyright (C) 2006 Canonical Ltd
 
1
# Copyright (C) 2006-2011 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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
16
 
17
17
import base64
18
 
import os
19
18
from StringIO import StringIO
 
19
import urlparse
20
20
import xmlrpclib
21
21
 
22
 
from bzrlib.tests import TestCase, TestSkipped
 
22
from bzrlib import (
 
23
    config,
 
24
    tests,
 
25
    ui,
 
26
    )
 
27
from bzrlib.tests import TestCaseWithTransport
23
28
 
24
29
# local import
25
 
from lp_registration import (
 
30
from bzrlib.plugins.launchpad.lp_registration import (
26
31
        BaseRequest,
27
32
        BranchBugLinkRequest,
28
33
        BranchRegistrationRequest,
 
34
        ResolveLaunchpadPathRequest,
29
35
        LaunchpadService,
30
36
        )
31
37
 
53
59
        """
54
60
        return (200, 'OK', [])
55
61
 
 
62
    def getresponse(self, buffering=True):
 
63
        """Fake the http reply.
 
64
 
 
65
        This is used when running on Python 2.7, where xmlrpclib uses
 
66
        httplib.HTTPConnection in a different way than before.
 
67
        """
 
68
        class FakeHttpResponse(object):
 
69
 
 
70
            def __init__(self, status, reason, body):
 
71
                self.status = status
 
72
                self.reason = reason
 
73
                self.body = body
 
74
 
 
75
            def read(self, size=-1):
 
76
                return self.body.read(size)
 
77
 
 
78
            def getheader(self, name, default):
 
79
                # We don't have headers
 
80
                return default
 
81
 
 
82
        return FakeHttpResponse(200, 'OK', self.getfile())
 
83
 
56
84
    def getfile(self):
57
85
        """Return a fake file containing the response content."""
58
86
        return StringIO('''\
74
102
    # Python 2.5's xmlrpclib looks for this.
75
103
    _use_datetime = False
76
104
 
77
 
    def __init__(self, testcase):
 
105
    def __init__(self, testcase, expect_auth):
78
106
        self.testcase = testcase
 
107
        self.expect_auth = expect_auth
 
108
        self._connection = (None, None)
79
109
 
80
110
    def make_connection(self, host):
81
111
        host, http_headers, x509 = self.get_host_info(host)
82
112
        test = self.testcase
83
113
        self.connected_host = host
84
 
        auth_hdrs = [v for k,v in http_headers if k == 'Authorization']
85
 
        assert len(auth_hdrs) == 1
86
 
        authinfo = auth_hdrs[0]
87
 
        expected_auth = 'testuser@launchpad.net:testpassword'
88
 
        test.assertEquals(authinfo,
89
 
                'Basic ' + base64.encodestring(expected_auth).strip())
 
114
        if self.expect_auth:
 
115
            auth_hdrs = [v for k,v in http_headers if k == 'Authorization']
 
116
            if len(auth_hdrs) != 1:
 
117
                raise AssertionError("multiple auth headers: %r"
 
118
                    % (auth_hdrs,))
 
119
            authinfo = auth_hdrs[0]
 
120
            expected_auth = 'testuser@launchpad.net:testpassword'
 
121
            test.assertEquals(authinfo,
 
122
                    'Basic ' + base64.encodestring(expected_auth).strip())
 
123
        elif http_headers:
 
124
            raise AssertionError()
90
125
        return InstrumentedXMLRPCConnection(test)
91
126
 
92
127
    def send_request(self, connection, handler_path, request_body):
103
138
 
104
139
    def send_content(self, conn, request_body):
105
140
        unpacked, method = xmlrpclib.loads(request_body)
106
 
        assert None not in unpacked, \
107
 
                "xmlrpc result %r shouldn't contain None" % (unpacked,)
 
141
        if None in unpacked:
 
142
            raise AssertionError(
 
143
                "xmlrpc result %r shouldn't contain None" % (unpacked,))
108
144
        self.sent_params = unpacked
109
145
 
110
146
 
111
147
class MockLaunchpadService(LaunchpadService):
112
148
 
113
 
    def send_request(self, method_name, method_params):
 
149
    def send_request(self, method_name, method_params, authenticated):
114
150
        """Stash away the method details rather than sending them to a real server"""
115
151
        self.called_method_name = method_name
116
152
        self.called_method_params = method_params
117
 
 
118
 
 
119
 
class TestBranchRegistration(TestCase):
120
 
    SAMPLE_URL = 'http://bazaar-vcs.org/bzr/bzr.dev/'
121
 
    SAMPLE_OWNER = 'jhacker@foo.com'
122
 
    SAMPLE_BRANCH_ID = 'bzr.dev'
 
153
        self.called_authenticated = authenticated
 
154
 
 
155
 
 
156
class TestBranchRegistration(TestCaseWithTransport):
123
157
 
124
158
    def setUp(self):
125
159
        super(TestBranchRegistration, self).setUp()
126
160
        # make sure we have a reproducible standard environment
127
 
        if 'BZR_LP_XMLRPC_URL' in os.environ:
128
 
            del os.environ['BZR_LP_XMLRPC_URL']
 
161
        self.overrideEnv('BZR_LP_XMLRPC_URL', None)
129
162
 
130
163
    def test_register_help(self):
131
164
        """register-branch accepts --help"""
132
 
        out, err = self.run_bzr('register-branch', '--help')
 
165
        out, err = self.run_bzr(['register-branch', '--help'])
133
166
        self.assertContainsRe(out, r'Register a branch')
134
167
 
135
 
    def test_register_no_url(self):
 
168
    def test_register_no_url_no_branch(self):
136
169
        """register-branch command requires parameters"""
137
 
        self.run_bzr('register-branch', retcode=3)
 
170
        self.make_repository('.')
 
171
        self.run_bzr_error(
 
172
            ['register-branch requires a public branch url - '
 
173
             'see bzr help register-branch'],
 
174
            'register-branch')
 
175
 
 
176
    def test_register_no_url_in_published_branch_no_error(self):
 
177
        b = self.make_branch('.')
 
178
        b.set_public_branch('http://test-server.com/bzr/branch')
 
179
        out, err = self.run_bzr(['register-branch', '--dry-run'])
 
180
        self.assertEqual('Branch registered.\n', out)
 
181
        self.assertEqual('', err)
 
182
 
 
183
    def test_register_no_url_in_unpublished_branch_errors(self):
 
184
        b = self.make_branch('.')
 
185
        out, err = self.run_bzr_error(['no public branch'],
 
186
            ['register-branch', '--dry-run'])
 
187
        self.assertEqual('', out)
138
188
 
139
189
    def test_register_dry_run(self):
140
 
        out, err = self.run_bzr('register-branch',
 
190
        out, err = self.run_bzr(['register-branch',
141
191
                                'http://test-server.com/bzr/branch',
142
 
                                '--dry-run')
 
192
                                '--dry-run'])
143
193
        self.assertEquals(out, 'Branch registered.\n')
144
194
 
145
195
    def test_onto_transport(self):
146
 
        """Test how the request is sent by transmitting across a mock Transport"""
 
196
        """How the request is sent by transmitting across a mock Transport"""
147
197
        # use a real transport, but intercept at the http/xml layer
148
 
        transport = InstrumentedXMLRPCTransport(self)
 
198
        transport = InstrumentedXMLRPCTransport(self, expect_auth=True)
149
199
        service = LaunchpadService(transport)
150
200
        service.registrant_email = 'testuser@launchpad.net'
151
201
        service.registrant_password = 'testpassword'
167
217
                 'product'))
168
218
        self.assertTrue(transport.got_request)
169
219
 
 
220
    def test_onto_transport_unauthenticated(self):
 
221
        """An unauthenticated request is transmitted across a mock Transport"""
 
222
        transport = InstrumentedXMLRPCTransport(self, expect_auth=False)
 
223
        service = LaunchpadService(transport)
 
224
        resolve = ResolveLaunchpadPathRequest('bzr')
 
225
        resolve.submit(service)
 
226
        self.assertEquals(transport.connected_host, 'xmlrpc.launchpad.net')
 
227
        self.assertEquals(len(transport.sent_params), 1)
 
228
        self.assertEquals(transport.sent_params, ('bzr', ))
 
229
        self.assertTrue(transport.got_request)
 
230
 
170
231
    def test_subclass_request(self):
171
232
        """Define a new type of xmlrpc request"""
172
233
        class DummyRequest(BaseRequest):
186
247
        """Send registration to mock server"""
187
248
        test_case = self
188
249
        class MockRegistrationService(MockLaunchpadService):
189
 
            def send_request(self, method_name, method_params):
 
250
            def send_request(self, method_name, method_params, authenticated):
190
251
                test_case.assertEquals(method_name, "register_branch")
191
252
                test_case.assertEquals(list(method_params),
192
253
                        ['url', 'name', 'title', 'description', 'email', 'name'])
 
254
                test_case.assertEquals(authenticated, True)
193
255
                return 'result'
194
256
        service = MockRegistrationService()
195
257
        rego = BranchRegistrationRequest('url', 'name', 'title',
201
263
        """Send registration to mock server"""
202
264
        test_case = self
203
265
        class MockRegistrationService(MockLaunchpadService):
204
 
            def send_request(self, method_name, method_params):
 
266
            def send_request(self, method_name, method_params, authenticated):
205
267
                test_case.assertEquals(method_name, "register_branch")
206
268
                test_case.assertEquals(list(method_params),
207
269
                        ['http://server/branch', 'branch', '', '', '', ''])
 
270
                test_case.assertEquals(authenticated, True)
208
271
                return 'result'
209
272
        service = MockRegistrationService()
210
273
        rego = BranchRegistrationRequest('http://server/branch')
215
278
        """Send bug-branch link to mock server"""
216
279
        test_case = self
217
280
        class MockService(MockLaunchpadService):
218
 
            def send_request(self, method_name, method_params):
 
281
            def send_request(self, method_name, method_params, authenticated):
219
282
                test_case.assertEquals(method_name, "link_branch_to_bug")
220
283
                test_case.assertEquals(list(method_params),
221
284
                        ['http://server/branch', 1234, ''])
 
285
                test_case.assertEquals(authenticated, True)
222
286
                return 'http://launchpad.net/bug/1234'
223
287
        service = MockService()
224
288
        rego = BranchBugLinkRequest('http://server/branch', 1234)
225
289
        result = rego.submit(service)
226
290
        self.assertEquals(result, 'http://launchpad.net/bug/1234')
 
291
 
 
292
    def test_mock_resolve_lp_url(self):
 
293
        test_case = self
 
294
        class MockService(MockLaunchpadService):
 
295
            def send_request(self, method_name, method_params, authenticated):
 
296
                test_case.assertEquals(method_name, "resolve_lp_path")
 
297
                test_case.assertEquals(list(method_params), ['bzr'])
 
298
                test_case.assertEquals(authenticated, False)
 
299
                return dict(urls=[
 
300
                        'bzr+ssh://bazaar.launchpad.net~bzr/bzr/trunk',
 
301
                        'sftp://bazaar.launchpad.net~bzr/bzr/trunk',
 
302
                        'bzr+http://bazaar.launchpad.net~bzr/bzr/trunk',
 
303
                        'http://bazaar.launchpad.net~bzr/bzr/trunk'])
 
304
        service = MockService()
 
305
        resolve = ResolveLaunchpadPathRequest('bzr')
 
306
        result = resolve.submit(service)
 
307
        self.assertTrue('urls' in result)
 
308
        self.assertEquals(result['urls'], [
 
309
                'bzr+ssh://bazaar.launchpad.net~bzr/bzr/trunk',
 
310
                'sftp://bazaar.launchpad.net~bzr/bzr/trunk',
 
311
                'bzr+http://bazaar.launchpad.net~bzr/bzr/trunk',
 
312
                'http://bazaar.launchpad.net~bzr/bzr/trunk'])
 
313
 
 
314
 
 
315
class TestGatherUserCredentials(tests.TestCaseInTempDir):
 
316
 
 
317
    def setUp(self):
 
318
        super(TestGatherUserCredentials, self).setUp()
 
319
        # make sure we have a reproducible standard environment
 
320
        self.overrideEnv('BZR_LP_XMLRPC_URL', None)
 
321
 
 
322
    def test_gather_user_credentials_has_password(self):
 
323
        service = LaunchpadService()
 
324
        service.registrant_password = 'mypassword'
 
325
        # This should be a basic no-op, since we already have the password
 
326
        service.gather_user_credentials()
 
327
        self.assertEqual('mypassword', service.registrant_password)
 
328
 
 
329
    def test_gather_user_credentials_from_auth_conf(self):
 
330
        auth_path = config.authentication_config_filename()
 
331
        service = LaunchpadService()
 
332
        g_conf = config.GlobalStack()
 
333
        g_conf.set('email', 'Test User <test@user.com>')
 
334
        g_conf.store.save()
 
335
        # FIXME: auth_path base dir exists only because bazaar.conf has just
 
336
        # been saved, brittle... -- vila 20120731
 
337
        f = open(auth_path, 'wb')
 
338
        try:
 
339
            scheme, hostinfo = urlparse.urlsplit(service.service_url)[:2]
 
340
            f.write('[section]\n'
 
341
                    'scheme=%s\n'
 
342
                    'host=%s\n'
 
343
                    'user=test@user.com\n'
 
344
                    'password=testpass\n'
 
345
                    % (scheme, hostinfo))
 
346
        finally:
 
347
            f.close()
 
348
        self.assertIs(None, service.registrant_password)
 
349
        service.gather_user_credentials()
 
350
        self.assertEqual('test@user.com', service.registrant_email)
 
351
        self.assertEqual('testpass', service.registrant_password)
 
352
 
 
353
    def test_gather_user_credentials_prompts(self):
 
354
        service = LaunchpadService()
 
355
        self.assertIs(None, service.registrant_password)
 
356
        g_conf = config.GlobalStack()
 
357
        g_conf.set('email', 'Test User <test@user.com>')
 
358
        g_conf.store.save()
 
359
        stdout = tests.StringIOWrapper()
 
360
        stderr = tests.StringIOWrapper()
 
361
        ui.ui_factory = tests.TestUIFactory(stdin='userpass\n',
 
362
                                            stdout=stdout, stderr=stderr)
 
363
        self.assertIs(None, service.registrant_password)
 
364
        service.gather_user_credentials()
 
365
        self.assertEqual('test@user.com', service.registrant_email)
 
366
        self.assertEqual('userpass', service.registrant_password)
 
367
        self.assertEquals('', stdout.getvalue())
 
368
        self.assertContainsRe(stderr.getvalue(),
 
369
                             'launchpad.net password for test@user\\.com')
 
370