← 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: Aaron Bentley
  • Date: 2009-03-10 04:55:49 UTC
  • mto: This revision was merged to the branch mainline in revision 4112.
  • Revision ID: aaron@aaronbentley.com-20090310045549-j5jmgq190872oem7
Remove locking decorator

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/plugins/launchpad/test_register.py
1
 
# Copyright (C) 2006 by 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
17
17
import base64
18
18
import os
19
19
from StringIO import StringIO
 
20
import urlparse
20
21
import xmlrpclib
21
22
 
22
 
from bzrlib.tests import TestCase, TestSkipped
 
23
from bzrlib import (
 
24
    config,
 
25
    osutils,
 
26
    tests,
 
27
    ui,
 
28
    )
 
29
from bzrlib.tests import TestCaseWithTransport, TestSkipped
23
30
 
24
31
# local import
25
 
from lp_registration import (
 
32
from bzrlib.plugins.launchpad.lp_registration import (
26
33
        BaseRequest,
27
34
        BranchBugLinkRequest,
28
35
        BranchRegistrationRequest,
 
36
        ResolveLaunchpadPathRequest,
29
37
        LaunchpadService,
30
38
        )
31
39
 
74
82
    # Python 2.5's xmlrpclib looks for this.
75
83
    _use_datetime = False
76
84
 
77
 
    def __init__(self, testcase):
 
85
    def __init__(self, testcase, expect_auth):
78
86
        self.testcase = testcase
 
87
        self.expect_auth = expect_auth
79
88
 
80
89
    def make_connection(self, host):
81
90
        host, http_headers, x509 = self.get_host_info(host)
82
91
        test = self.testcase
83
92
        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())
 
93
        if self.expect_auth:
 
94
            auth_hdrs = [v for k,v in http_headers if k == 'Authorization']
 
95
            if len(auth_hdrs) != 1:
 
96
                raise AssertionError("multiple auth headers: %r"
 
97
                    % (auth_hdrs,))
 
98
            authinfo = auth_hdrs[0]
 
99
            expected_auth = 'testuser@launchpad.net:testpassword'
 
100
            test.assertEquals(authinfo,
 
101
                    'Basic ' + base64.encodestring(expected_auth).strip())
 
102
        elif http_headers:
 
103
            raise AssertionError()
90
104
        return InstrumentedXMLRPCConnection(test)
91
105
 
92
106
    def send_request(self, connection, handler_path, request_body):
103
117
 
104
118
    def send_content(self, conn, request_body):
105
119
        unpacked, method = xmlrpclib.loads(request_body)
106
 
        assert None not in unpacked, \
107
 
                "xmlrpc result %r shouldn't contain None" % (unpacked,)
 
120
        if None in unpacked:
 
121
            raise AssertionError(
 
122
                "xmlrpc result %r shouldn't contain None" % (unpacked,))
108
123
        self.sent_params = unpacked
109
124
 
110
125
 
111
126
class MockLaunchpadService(LaunchpadService):
112
127
 
113
 
    def send_request(self, method_name, method_params):
 
128
    def send_request(self, method_name, method_params, authenticated):
114
129
        """Stash away the method details rather than sending them to a real server"""
115
130
        self.called_method_name = method_name
116
131
        self.called_method_params = method_params
117
 
 
118
 
 
119
 
class TestBranchRegistration(TestCase):
 
132
        self.called_authenticated = authenticated
 
133
 
 
134
 
 
135
class TestBranchRegistration(TestCaseWithTransport):
120
136
    SAMPLE_URL = 'http://bazaar-vcs.org/bzr/bzr.dev/'
121
137
    SAMPLE_OWNER = 'jhacker@foo.com'
122
138
    SAMPLE_BRANCH_ID = 'bzr.dev'
124
140
    def setUp(self):
125
141
        super(TestBranchRegistration, self).setUp()
126
142
        # 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']
 
143
        self._captureVar('BZR_LP_XMLRPC_URL', None)
129
144
 
130
145
    def test_register_help(self):
131
146
        """register-branch accepts --help"""
132
 
        out, err = self.run_bzr('register-branch', '--help')
 
147
        out, err = self.run_bzr(['register-branch', '--help'])
133
148
        self.assertContainsRe(out, r'Register a branch')
134
149
 
135
 
    def test_register_no_url(self):
 
150
    def test_register_no_url_no_branch(self):
136
151
        """register-branch command requires parameters"""
137
 
        self.run_bzr('register-branch', retcode=3)
 
152
        self.make_repository('.')
 
153
        self.run_bzr_error(
 
154
            ['register-branch requires a public branch url - '
 
155
             'see bzr help register-branch'],
 
156
            'register-branch')
 
157
 
 
158
    def test_register_no_url_in_published_branch_no_error(self):
 
159
        b = self.make_branch('.')
 
160
        b.set_public_branch('http://test-server.com/bzr/branch')
 
161
        out, err = self.run_bzr(['register-branch', '--dry-run'])
 
162
        self.assertEqual('Branch registered.\n', out)
 
163
        self.assertEqual('', err)
 
164
 
 
165
    def test_register_no_url_in_unpublished_branch_errors(self):
 
166
        b = self.make_branch('.')
 
167
        out, err = self.run_bzr_error(['no public branch'],
 
168
            ['register-branch', '--dry-run'])
 
169
        self.assertEqual('', out)
138
170
 
139
171
    def test_register_dry_run(self):
140
 
        out, err = self.run_bzr('register-branch',
 
172
        out, err = self.run_bzr(['register-branch',
141
173
                                'http://test-server.com/bzr/branch',
142
 
                                '--dry-run')
 
174
                                '--dry-run'])
143
175
        self.assertEquals(out, 'Branch registered.\n')
144
176
 
145
177
    def test_onto_transport(self):
146
178
        """Test how the request is sent by transmitting across a mock Transport"""
147
179
        # use a real transport, but intercept at the http/xml layer
148
 
        transport = InstrumentedXMLRPCTransport(self)
 
180
        transport = InstrumentedXMLRPCTransport(self, expect_auth=True)
149
181
        service = LaunchpadService(transport)
150
182
        service.registrant_email = 'testuser@launchpad.net'
151
183
        service.registrant_password = 'testpassword'
156
188
                'author@launchpad.net',
157
189
                'product')
158
190
        rego.submit(service)
159
 
        self.assertEquals(transport.connected_host, 'xmlrpc.launchpad.net')
 
191
        self.assertEquals(transport.connected_host, 'xmlrpc.edge.launchpad.net')
160
192
        self.assertEquals(len(transport.sent_params), 6)
161
193
        self.assertEquals(transport.sent_params,
162
194
                ('http://test-server.com/bzr/branch',  # branch_url
167
199
                 'product'))
168
200
        self.assertTrue(transport.got_request)
169
201
 
 
202
    def test_onto_transport_unauthenticated(self):
 
203
        """Test how an unauthenticated request is transmitted across a mock Transport"""
 
204
        transport = InstrumentedXMLRPCTransport(self, expect_auth=False)
 
205
        service = LaunchpadService(transport)
 
206
        resolve = ResolveLaunchpadPathRequest('bzr')
 
207
        resolve.submit(service)
 
208
        self.assertEquals(transport.connected_host, 'xmlrpc.edge.launchpad.net')
 
209
        self.assertEquals(len(transport.sent_params), 1)
 
210
        self.assertEquals(transport.sent_params, ('bzr', ))
 
211
        self.assertTrue(transport.got_request)
 
212
 
170
213
    def test_subclass_request(self):
171
214
        """Define a new type of xmlrpc request"""
172
215
        class DummyRequest(BaseRequest):
186
229
        """Send registration to mock server"""
187
230
        test_case = self
188
231
        class MockRegistrationService(MockLaunchpadService):
189
 
            def send_request(self, method_name, method_params):
 
232
            def send_request(self, method_name, method_params, authenticated):
190
233
                test_case.assertEquals(method_name, "register_branch")
191
234
                test_case.assertEquals(list(method_params),
192
235
                        ['url', 'name', 'title', 'description', 'email', 'name'])
 
236
                test_case.assertEquals(authenticated, True)
193
237
                return 'result'
194
238
        service = MockRegistrationService()
195
239
        rego = BranchRegistrationRequest('url', 'name', 'title',
201
245
        """Send registration to mock server"""
202
246
        test_case = self
203
247
        class MockRegistrationService(MockLaunchpadService):
204
 
            def send_request(self, method_name, method_params):
 
248
            def send_request(self, method_name, method_params, authenticated):
205
249
                test_case.assertEquals(method_name, "register_branch")
206
250
                test_case.assertEquals(list(method_params),
207
251
                        ['http://server/branch', 'branch', '', '', '', ''])
 
252
                test_case.assertEquals(authenticated, True)
208
253
                return 'result'
209
254
        service = MockRegistrationService()
210
255
        rego = BranchRegistrationRequest('http://server/branch')
215
260
        """Send bug-branch link to mock server"""
216
261
        test_case = self
217
262
        class MockService(MockLaunchpadService):
218
 
            def send_request(self, method_name, method_params):
 
263
            def send_request(self, method_name, method_params, authenticated):
219
264
                test_case.assertEquals(method_name, "link_branch_to_bug")
220
265
                test_case.assertEquals(list(method_params),
221
266
                        ['http://server/branch', 1234, ''])
 
267
                test_case.assertEquals(authenticated, True)
222
268
                return 'http://launchpad.net/bug/1234'
223
269
        service = MockService()
224
270
        rego = BranchBugLinkRequest('http://server/branch', 1234)
225
271
        result = rego.submit(service)
226
272
        self.assertEquals(result, 'http://launchpad.net/bug/1234')
 
273
 
 
274
    def test_mock_resolve_lp_url(self):
 
275
        test_case = self
 
276
        class MockService(MockLaunchpadService):
 
277
            def send_request(self, method_name, method_params, authenticated):
 
278
                test_case.assertEquals(method_name, "resolve_lp_path")
 
279
                test_case.assertEquals(list(method_params), ['bzr'])
 
280
                test_case.assertEquals(authenticated, False)
 
281
                return dict(urls=[
 
282
                        'bzr+ssh://bazaar.launchpad.net~bzr/bzr/trunk',
 
283
                        'sftp://bazaar.launchpad.net~bzr/bzr/trunk',
 
284
                        'bzr+http://bazaar.launchpad.net~bzr/bzr/trunk',
 
285
                        'http://bazaar.launchpad.net~bzr/bzr/trunk'])
 
286
        service = MockService()
 
287
        resolve = ResolveLaunchpadPathRequest('bzr')
 
288
        result = resolve.submit(service)
 
289
        self.assertTrue('urls' in result)
 
290
        self.assertEquals(result['urls'], [
 
291
                'bzr+ssh://bazaar.launchpad.net~bzr/bzr/trunk',
 
292
                'sftp://bazaar.launchpad.net~bzr/bzr/trunk',
 
293
                'bzr+http://bazaar.launchpad.net~bzr/bzr/trunk',
 
294
                'http://bazaar.launchpad.net~bzr/bzr/trunk'])
 
295
 
 
296
 
 
297
class TestGatherUserCredentials(tests.TestCaseInTempDir):
 
298
 
 
299
    def setUp(self):
 
300
        super(TestGatherUserCredentials, self).setUp()
 
301
        # make sure we have a reproducible standard environment
 
302
        self._captureVar('BZR_LP_XMLRPC_URL', None)
 
303
 
 
304
    def test_gather_user_credentials_has_password(self):
 
305
        service = LaunchpadService()
 
306
        service.registrant_password = 'mypassword'
 
307
        # This should be a basic no-op, since we already have the password
 
308
        service.gather_user_credentials()
 
309
        self.assertEqual('mypassword', service.registrant_password)
 
310
 
 
311
    def test_gather_user_credentials_from_auth_conf(self):
 
312
        auth_path = config.authentication_config_filename()
 
313
        service = LaunchpadService()
 
314
        g_conf = config.GlobalConfig()
 
315
        g_conf.set_user_option('email', 'Test User <test@user.com>')
 
316
        f = open(auth_path, 'wb')
 
317
        try:
 
318
            scheme, hostinfo = urlparse.urlsplit(service.service_url)[:2]
 
319
            f.write('[section]\n'
 
320
                    'scheme=%s\n'
 
321
                    'host=%s\n'
 
322
                    'user=test@user.com\n'
 
323
                    'password=testpass\n'
 
324
                    % (scheme, hostinfo))
 
325
        finally:
 
326
            f.close()
 
327
        self.assertIs(None, service.registrant_password)
 
328
        service.gather_user_credentials()
 
329
        self.assertEqual('test@user.com', service.registrant_email)
 
330
        self.assertEqual('testpass', service.registrant_password)
 
331
 
 
332
    def test_gather_user_credentials_prompts(self):
 
333
        service = LaunchpadService()
 
334
        self.assertIs(None, service.registrant_password)
 
335
        g_conf = config.GlobalConfig()
 
336
        g_conf.set_user_option('email', 'Test User <test@user.com>')
 
337
        stdout = tests.StringIOWrapper()
 
338
        ui.ui_factory = tests.TestUIFactory(stdin='userpass\n',
 
339
                                            stdout=stdout)
 
340
        self.assertIs(None, service.registrant_password)
 
341
        service.gather_user_credentials()
 
342
        self.assertEqual('test@user.com', service.registrant_email)
 
343
        self.assertEqual('userpass', service.registrant_password)
 
344
        self.assertContainsRe(stdout.getvalue(),
 
345
                             'launchpad.net password for test@user\\.com')
 
346