1
# Copyright (C) 2006-2011 Canonical Ltd
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.
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
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
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
18
from StringIO import StringIO
27
from bzrlib.tests import TestCaseWithTransport
30
from bzrlib.plugins.launchpad.lp_registration import (
33
BranchRegistrationRequest,
34
ResolveLaunchpadPathRequest,
39
# TODO: Test that the command-line client, making sure that it'll pass the
40
# request through to a dummy transport, and that the transport will validate
41
# the results passed in. Not sure how to get the transport object back out to
42
# validate that its OK - may not be necessary.
44
# TODO: Add test for (and implement) other command-line options to set
45
# project, author_email, description.
47
# TODO: project_id is not properly handled -- must be passed in rpc or path.
49
class InstrumentedXMLRPCConnection(object):
50
"""Stands in place of an http connection for the purposes of testing"""
52
def __init__(self, testcase):
53
self.testcase = testcase
56
"""Fake the http reply.
58
:returns: (errcode, errmsg, headers)
60
return (200, 'OK', [])
62
def getresponse(self, buffering=True):
63
"""Fake the http reply.
65
This is used when running on Python 2.7, where xmlrpclib uses
66
httplib.HTTPConnection in a different way than before.
68
class FakeHttpResponse(object):
70
def __init__(self, status, reason, body):
75
def read(self, size=-1):
76
return self.body.read(size)
78
def getheader(self, name, default):
79
# We don't have headers
82
return FakeHttpResponse(200, 'OK', self.getfile())
85
"""Return a fake file containing the response content."""
87
<?xml version="1.0" ?>
92
<string>victoria dock</string>
100
class InstrumentedXMLRPCTransport(xmlrpclib.Transport):
102
# Python 2.5's xmlrpclib looks for this.
103
_use_datetime = False
105
def __init__(self, testcase, expect_auth):
106
self.testcase = testcase
107
self.expect_auth = expect_auth
108
self._connection = (None, None)
110
def make_connection(self, host):
111
host, http_headers, x509 = self.get_host_info(host)
113
self.connected_host = host
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"
119
authinfo = auth_hdrs[0]
120
expected_auth = 'testuser@launchpad.net:testpassword'
121
test.assertEquals(authinfo,
122
'Basic ' + base64.encodestring(expected_auth).strip())
124
raise AssertionError()
125
return InstrumentedXMLRPCConnection(test)
127
def send_request(self, connection, handler_path, request_body):
129
self.got_request = True
131
def send_host(self, conn, host):
134
def send_user_agent(self, conn):
135
# TODO: send special user agent string, including bzrlib version
139
def send_content(self, conn, request_body):
140
unpacked, method = xmlrpclib.loads(request_body)
142
raise AssertionError(
143
"xmlrpc result %r shouldn't contain None" % (unpacked,))
144
self.sent_params = unpacked
147
class MockLaunchpadService(LaunchpadService):
149
def send_request(self, method_name, method_params, authenticated):
150
"""Stash away the method details rather than sending them to a real server"""
151
self.called_method_name = method_name
152
self.called_method_params = method_params
153
self.called_authenticated = authenticated
156
class TestBranchRegistration(TestCaseWithTransport):
159
super(TestBranchRegistration, self).setUp()
160
# make sure we have a reproducible standard environment
161
self.overrideEnv('BZR_LP_XMLRPC_URL', None)
163
def test_register_help(self):
164
"""register-branch accepts --help"""
165
out, err = self.run_bzr(['register-branch', '--help'])
166
self.assertContainsRe(out, r'Register a branch')
168
def test_register_no_url_no_branch(self):
169
"""register-branch command requires parameters"""
170
self.make_repository('.')
172
['register-branch requires a public branch url - '
173
'see bzr help register-branch'],
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)
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)
189
def test_register_dry_run(self):
190
out, err = self.run_bzr(['register-branch',
191
'http://test-server.com/bzr/branch',
193
self.assertEquals(out, 'Branch registered.\n')
195
def test_onto_transport(self):
196
"""How the request is sent by transmitting across a mock Transport"""
197
# use a real transport, but intercept at the http/xml layer
198
transport = InstrumentedXMLRPCTransport(self, expect_auth=True)
199
service = LaunchpadService(transport)
200
service.registrant_email = 'testuser@launchpad.net'
201
service.registrant_password = 'testpassword'
202
rego = BranchRegistrationRequest('http://test-server.com/bzr/branch',
206
'author@launchpad.net',
209
self.assertEquals(transport.connected_host, 'xmlrpc.launchpad.net')
210
self.assertEquals(len(transport.sent_params), 6)
211
self.assertEquals(transport.sent_params,
212
('http://test-server.com/bzr/branch', # branch_url
213
'branch-id', # branch_name
214
'my test branch', # branch_title
216
'author@launchpad.net',
218
self.assertTrue(transport.got_request)
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)
231
def test_subclass_request(self):
232
"""Define a new type of xmlrpc request"""
233
class DummyRequest(BaseRequest):
234
_methodname = 'dummy_request'
235
def _request_params(self):
238
service = MockLaunchpadService()
239
service.registrant_email = 'test@launchpad.net'
240
service.registrant_password = ''
241
request = DummyRequest()
242
request.submit(service)
243
self.assertEquals(service.called_method_name, 'dummy_request')
244
self.assertEquals(service.called_method_params, (42,))
246
def test_mock_server_registration(self):
247
"""Send registration to mock server"""
249
class MockRegistrationService(MockLaunchpadService):
250
def send_request(self, method_name, method_params, authenticated):
251
test_case.assertEquals(method_name, "register_branch")
252
test_case.assertEquals(list(method_params),
253
['url', 'name', 'title', 'description', 'email', 'name'])
254
test_case.assertEquals(authenticated, True)
256
service = MockRegistrationService()
257
rego = BranchRegistrationRequest('url', 'name', 'title',
258
'description', 'email', 'name')
259
result = rego.submit(service)
260
self.assertEquals(result, 'result')
262
def test_mock_server_registration_with_defaults(self):
263
"""Send registration to mock server"""
265
class MockRegistrationService(MockLaunchpadService):
266
def send_request(self, method_name, method_params, authenticated):
267
test_case.assertEquals(method_name, "register_branch")
268
test_case.assertEquals(list(method_params),
269
['http://server/branch', 'branch', '', '', '', ''])
270
test_case.assertEquals(authenticated, True)
272
service = MockRegistrationService()
273
rego = BranchRegistrationRequest('http://server/branch')
274
result = rego.submit(service)
275
self.assertEquals(result, 'result')
277
def test_mock_bug_branch_link(self):
278
"""Send bug-branch link to mock server"""
280
class MockService(MockLaunchpadService):
281
def send_request(self, method_name, method_params, authenticated):
282
test_case.assertEquals(method_name, "link_branch_to_bug")
283
test_case.assertEquals(list(method_params),
284
['http://server/branch', 1234, ''])
285
test_case.assertEquals(authenticated, True)
286
return 'http://launchpad.net/bug/1234'
287
service = MockService()
288
rego = BranchBugLinkRequest('http://server/branch', 1234)
289
result = rego.submit(service)
290
self.assertEquals(result, 'http://launchpad.net/bug/1234')
292
def test_mock_resolve_lp_url(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)
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'])
315
class TestGatherUserCredentials(tests.TestCaseInTempDir):
318
super(TestGatherUserCredentials, self).setUp()
319
# make sure we have a reproducible standard environment
320
self.overrideEnv('BZR_LP_XMLRPC_URL', None)
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)
329
def test_gather_user_credentials_from_auth_conf(self):
330
auth_path = config.authentication_config_filename()
331
service = LaunchpadService()
332
g_conf = config.GlobalConfig()
333
g_conf.set_user_option('email', 'Test User <test@user.com>')
334
f = open(auth_path, 'wb')
336
scheme, hostinfo = urlparse.urlsplit(service.service_url)[:2]
337
f.write('[section]\n'
340
'user=test@user.com\n'
341
'password=testpass\n'
342
% (scheme, hostinfo))
345
self.assertIs(None, service.registrant_password)
346
service.gather_user_credentials()
347
self.assertEqual('test@user.com', service.registrant_email)
348
self.assertEqual('testpass', service.registrant_password)
350
def test_gather_user_credentials_prompts(self):
351
service = LaunchpadService()
352
self.assertIs(None, service.registrant_password)
353
g_conf = config.GlobalConfig()
354
g_conf.set_user_option('email', 'Test User <test@user.com>')
355
stdout = tests.StringIOWrapper()
356
stderr = tests.StringIOWrapper()
357
ui.ui_factory = tests.TestUIFactory(stdin='userpass\n',
358
stdout=stdout, stderr=stderr)
359
self.assertIs(None, service.registrant_password)
360
service.gather_user_credentials()
361
self.assertEqual('test@user.com', service.registrant_email)
362
self.assertEqual('userpass', service.registrant_password)
363
self.assertEquals('', stdout.getvalue())
364
self.assertContainsRe(stderr.getvalue(),
365
'launchpad.net password for test@user\\.com')
added
removed
bzrlib/plugins/launchpad/test_register.py
