diff options
Diffstat (limited to 'bzrlib/plugins/launchpad/test_register.py')
| -rw-r--r-- | bzrlib/plugins/launchpad/test_register.py | 366 |
1 files changed, 366 insertions, 0 deletions
diff --git a/bzrlib/plugins/launchpad/test_register.py b/bzrlib/plugins/launchpad/test_register.py new file mode 100644 index 0000000..81aadc7 --- /dev/null +++ b/bzrlib/plugins/launchpad/test_register.py @@ -0,0 +1,366 @@ +# Copyright (C) 2006-2011 Canonical Ltd +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA + +import base64 +from StringIO import StringIO +import urlparse +import xmlrpclib + +from bzrlib import ( + config, + tests, + ui, + ) +from bzrlib.tests import TestCaseWithTransport + +# local import +from bzrlib.plugins.launchpad.lp_registration import ( + BaseRequest, + BranchBugLinkRequest, + BranchRegistrationRequest, + ResolveLaunchpadPathRequest, + LaunchpadService, + ) + + +# TODO: Test that the command-line client, making sure that it'll pass the +# request through to a dummy transport, and that the transport will validate +# the results passed in. Not sure how to get the transport object back out to +# validate that its OK - may not be necessary. + +# TODO: Add test for (and implement) other command-line options to set +# project, author_email, description. + +# TODO: project_id is not properly handled -- must be passed in rpc or path. + +class InstrumentedXMLRPCConnection(object): + """Stands in place of an http connection for the purposes of testing""" + + def __init__(self, testcase): + self.testcase = testcase + + def getreply(self): + """Fake the http reply. + + :returns: (errcode, errmsg, headers) + """ + return (200, 'OK', []) + + def getresponse(self, buffering=True): + """Fake the http reply. + + This is used when running on Python 2.7, where xmlrpclib uses + httplib.HTTPConnection in a different way than before. + """ + class FakeHttpResponse(object): + + def __init__(self, status, reason, body): + self.status = status + self.reason = reason + self.body = body + + def read(self, size=-1): + return self.body.read(size) + + def getheader(self, name, default): + # We don't have headers + return default + + return FakeHttpResponse(200, 'OK', self.getfile()) + + def getfile(self): + """Return a fake file containing the response content.""" + return StringIO('''\ +<?xml version="1.0" ?> +<methodResponse> + <params> + <param> + <value> + <string>victoria dock</string> + </value> + </param> + </params> +</methodResponse>''') + + + +class InstrumentedXMLRPCTransport(xmlrpclib.Transport): + + # Python 2.5's xmlrpclib looks for this. + _use_datetime = False + + def __init__(self, testcase, expect_auth): + self.testcase = testcase + self.expect_auth = expect_auth + self._connection = (None, None) + + def make_connection(self, host): + host, http_headers, x509 = self.get_host_info(host) + test = self.testcase + self.connected_host = host + if self.expect_auth: + auth_hdrs = [v for k,v in http_headers if k == 'Authorization'] + if len(auth_hdrs) != 1: + raise AssertionError("multiple auth headers: %r" + % (auth_hdrs,)) + authinfo = auth_hdrs[0] + expected_auth = 'testuser@launchpad.net:testpassword' + test.assertEquals(authinfo, + 'Basic ' + base64.encodestring(expected_auth).strip()) + elif http_headers: + raise AssertionError() + return InstrumentedXMLRPCConnection(test) + + def send_request(self, connection, handler_path, request_body): + test = self.testcase + self.got_request = True + + def send_host(self, conn, host): + pass + + def send_user_agent(self, conn): + # TODO: send special user agent string, including bzrlib version + # number + pass + + def send_content(self, conn, request_body): + unpacked, method = xmlrpclib.loads(request_body) + if None in unpacked: + raise AssertionError( + "xmlrpc result %r shouldn't contain None" % (unpacked,)) + self.sent_params = unpacked + + +class MockLaunchpadService(LaunchpadService): + + def send_request(self, method_name, method_params, authenticated): + """Stash away the method details rather than sending them to a real server""" + self.called_method_name = method_name + self.called_method_params = method_params + self.called_authenticated = authenticated + + +class TestBranchRegistration(TestCaseWithTransport): + + def setUp(self): + super(TestBranchRegistration, self).setUp() + # make sure we have a reproducible standard environment + self.overrideEnv('BZR_LP_XMLRPC_URL', None) + + def test_register_help(self): + """register-branch accepts --help""" + out, err = self.run_bzr(['register-branch', '--help']) + self.assertContainsRe(out, r'Register a branch') + + def test_register_no_url_no_branch(self): + """register-branch command requires parameters""" + self.make_repository('.') + self.run_bzr_error( + ['register-branch requires a public branch url - ' + 'see bzr help register-branch'], + 'register-branch') + + def test_register_no_url_in_published_branch_no_error(self): + b = self.make_branch('.') + b.set_public_branch('http://test-server.com/bzr/branch') + out, err = self.run_bzr(['register-branch', '--dry-run']) + self.assertEqual('Branch registered.\n', out) + self.assertEqual('', err) + + def test_register_no_url_in_unpublished_branch_errors(self): + b = self.make_branch('.') + out, err = self.run_bzr_error(['no public branch'], + ['register-branch', '--dry-run']) + self.assertEqual('', out) + + def test_register_dry_run(self): + out, err = self.run_bzr(['register-branch', + 'http://test-server.com/bzr/branch', + '--dry-run']) + self.assertEquals(out, 'Branch registered.\n') + + def test_onto_transport(self): + """How the request is sent by transmitting across a mock Transport""" + # use a real transport, but intercept at the http/xml layer + transport = InstrumentedXMLRPCTransport(self, expect_auth=True) + service = LaunchpadService(transport) + service.registrant_email = 'testuser@launchpad.net' + service.registrant_password = 'testpassword' + rego = BranchRegistrationRequest('http://test-server.com/bzr/branch', + 'branch-id', + 'my test branch', + 'description', + 'author@launchpad.net', + 'product') + rego.submit(service) + self.assertEquals(transport.connected_host, 'xmlrpc.launchpad.net') + self.assertEquals(len(transport.sent_params), 6) + self.assertEquals(transport.sent_params, + ('http://test-server.com/bzr/branch', # branch_url + 'branch-id', # branch_name + 'my test branch', # branch_title + 'description', + 'author@launchpad.net', + 'product')) + self.assertTrue(transport.got_request) + + def test_onto_transport_unauthenticated(self): + """An unauthenticated request is transmitted across a mock Transport""" + transport = InstrumentedXMLRPCTransport(self, expect_auth=False) + service = LaunchpadService(transport) + resolve = ResolveLaunchpadPathRequest('bzr') + resolve.submit(service) + self.assertEquals(transport.connected_host, 'xmlrpc.launchpad.net') + self.assertEquals(len(transport.sent_params), 1) + self.assertEquals(transport.sent_params, ('bzr', )) + self.assertTrue(transport.got_request) + + def test_subclass_request(self): + """Define a new type of xmlrpc request""" + class DummyRequest(BaseRequest): + _methodname = 'dummy_request' + def _request_params(self): + return (42,) + + service = MockLaunchpadService() + service.registrant_email = 'test@launchpad.net' + service.registrant_password = '' + request = DummyRequest() + request.submit(service) + self.assertEquals(service.called_method_name, 'dummy_request') + self.assertEquals(service.called_method_params, (42,)) + + def test_mock_server_registration(self): + """Send registration to mock server""" + test_case = self + class MockRegistrationService(MockLaunchpadService): + def send_request(self, method_name, method_params, authenticated): + test_case.assertEquals(method_name, "register_branch") + test_case.assertEquals(list(method_params), + ['url', 'name', 'title', 'description', 'email', 'name']) + test_case.assertEquals(authenticated, True) + return 'result' + service = MockRegistrationService() + rego = BranchRegistrationRequest('url', 'name', 'title', + 'description', 'email', 'name') + result = rego.submit(service) + self.assertEquals(result, 'result') + + def test_mock_server_registration_with_defaults(self): + """Send registration to mock server""" + test_case = self + class MockRegistrationService(MockLaunchpadService): + def send_request(self, method_name, method_params, authenticated): + test_case.assertEquals(method_name, "register_branch") + test_case.assertEquals(list(method_params), + ['http://server/branch', 'branch', '', '', '', '']) + test_case.assertEquals(authenticated, True) + return 'result' + service = MockRegistrationService() + rego = BranchRegistrationRequest('http://server/branch') + result = rego.submit(service) + self.assertEquals(result, 'result') + + def test_mock_bug_branch_link(self): + """Send bug-branch link to mock server""" + test_case = self + class MockService(MockLaunchpadService): + def send_request(self, method_name, method_params, authenticated): + test_case.assertEquals(method_name, "link_branch_to_bug") + test_case.assertEquals(list(method_params), + ['http://server/branch', 1234, '']) + test_case.assertEquals(authenticated, True) + return 'http://launchpad.net/bug/1234' + service = MockService() + rego = BranchBugLinkRequest('http://server/branch', 1234) + result = rego.submit(service) + self.assertEquals(result, 'http://launchpad.net/bug/1234') + + def test_mock_resolve_lp_url(self): + test_case = self + class MockService(MockLaunchpadService): + def send_request(self, method_name, method_params, authenticated): + test_case.assertEquals(method_name, "resolve_lp_path") + test_case.assertEquals(list(method_params), ['bzr']) + test_case.assertEquals(authenticated, False) + return dict(urls=[ + 'bzr+ssh://bazaar.launchpad.net~bzr/bzr/trunk', + 'sftp://bazaar.launchpad.net~bzr/bzr/trunk', + 'bzr+http://bazaar.launchpad.net~bzr/bzr/trunk', + 'http://bazaar.launchpad.net~bzr/bzr/trunk']) + service = MockService() + resolve = ResolveLaunchpadPathRequest('bzr') + result = resolve.submit(service) + self.assertTrue('urls' in result) + self.assertEquals(result['urls'], [ + 'bzr+ssh://bazaar.launchpad.net~bzr/bzr/trunk', + 'sftp://bazaar.launchpad.net~bzr/bzr/trunk', + 'bzr+http://bazaar.launchpad.net~bzr/bzr/trunk', + 'http://bazaar.launchpad.net~bzr/bzr/trunk']) + + +class TestGatherUserCredentials(tests.TestCaseInTempDir): + + def setUp(self): + super(TestGatherUserCredentials, self).setUp() + # make sure we have a reproducible standard environment + self.overrideEnv('BZR_LP_XMLRPC_URL', None) + + def test_gather_user_credentials_has_password(self): + service = LaunchpadService() + service.registrant_password = 'mypassword' + # This should be a basic no-op, since we already have the password + service.gather_user_credentials() + self.assertEqual('mypassword', service.registrant_password) + + def test_gather_user_credentials_from_auth_conf(self): + auth_path = config.authentication_config_filename() + service = LaunchpadService() + g_conf = config.GlobalStack() + g_conf.set('email', 'Test User <test@user.com>') + f = open(auth_path, 'wb') + try: + scheme, hostinfo = urlparse.urlsplit(service.service_url)[:2] + f.write('[section]\n' + 'scheme=%s\n' + 'host=%s\n' + 'user=test@user.com\n' + 'password=testpass\n' + % (scheme, hostinfo)) + finally: + f.close() + self.assertIs(None, service.registrant_password) + service.gather_user_credentials() + self.assertEqual('test@user.com', service.registrant_email) + self.assertEqual('testpass', service.registrant_password) + + def test_gather_user_credentials_prompts(self): + service = LaunchpadService() + self.assertIs(None, service.registrant_password) + g_conf = config.GlobalStack() + g_conf.set('email', 'Test User <test@user.com>') + stdout = tests.StringIOWrapper() + stderr = tests.StringIOWrapper() + ui.ui_factory = tests.TestUIFactory(stdin='userpass\n', + stdout=stdout, stderr=stderr) + self.assertIs(None, service.registrant_password) + service.gather_user_credentials() + self.assertEqual('test@user.com', service.registrant_email) + self.assertEqual('userpass', service.registrant_password) + self.assertEquals('', stdout.getvalue()) + self.assertContainsRe(stderr.getvalue(), + 'launchpad.net password for test@user\\.com') + |