Rev 4778: Start testing the XMLRPC transport re-implemented on top of _urllib2_wrappers. in file:///home/vila/src/bzr/bugs/186920-lp-proxy/
Vincent Ladeuil
v.ladeuil+lp at free.fr
Fri Oct 30 14:44:52 GMT 2009
At file:///home/vila/src/bzr/bugs/186920-lp-proxy/
------------------------------------------------------------
revno: 4778
revision-id: v.ladeuil+lp at free.fr-20091030144452-f5pqbo4kwqrb1row
parent: v.ladeuil+lp at free.fr-20091030093450-vubsvcm9b16o0ask
committer: Vincent Ladeuil <v.ladeuil+lp at free.fr>
branch nick: 186920-lp-proxy
timestamp: Fri 2009-10-30 15:44:52 +0100
message:
Start testing the XMLRPC transport re-implemented on top of _urllib2_wrappers.
* bzrlib/plugins/launchpad/test_lp_directory.py:
(load_tests): Add some parameterized tests so we exercise both the
http and https transports.
(TestTransport): Test the transport stack against a predefined
server.
* bzrlib/plugins/launchpad/lp_registration.py:
(Transport): Implement the RPC calls on top of _urllib2_wrappers
so we can take care of proxies.
* bzrlib/plugins/launchpad/__init__.py:
(load_tests): Use the modern way to load tests.
-------------- next part --------------
=== modified file 'bzrlib/plugins/launchpad/__init__.py'
--- a/bzrlib/plugins/launchpad/__init__.py 2009-07-06 13:00:23 +0000
+++ b/bzrlib/plugins/launchpad/__init__.py 2009-10-30 14:44:52 +0000
@@ -262,30 +262,19 @@
_register_directory()
-def test_suite():
- """Called by bzrlib to fetch tests for this plugin"""
- from unittest import TestSuite, TestLoader
- from bzrlib.plugins.launchpad import (
- test_account,
- test_lp_directory,
- test_lp_login,
- test_lp_open,
- test_lp_service,
- test_register,
- )
+def load_tests(basic_tests, module, loader):
+ testmod_names = [
+ 'test_account',
+ 'test_register',
+ 'test_lp_directory',
+ 'test_lp_login',
+ 'test_lp_open',
+ 'test_lp_service',
+ ]
+ basic_tests.addTest(loader.loadTestsFromModuleNames(
+ ["%s.%s" % (__name__, tmn) for tmn in testmod_names]))
+ return basic_tests
- loader = TestLoader()
- suite = TestSuite()
- for module in [
- test_account,
- test_register,
- test_lp_directory,
- test_lp_login,
- test_lp_open,
- test_lp_service,
- ]:
- suite.addTests(loader.loadTestsFromModule(module))
- return suite
_launchpad_help = """Integration with Launchpad.net
=== modified file 'bzrlib/plugins/launchpad/lp_registration.py'
--- a/bzrlib/plugins/launchpad/lp_registration.py 2009-07-04 16:22:16 +0000
+++ b/bzrlib/plugins/launchpad/lp_registration.py 2009-10-30 14:44:52 +0000
@@ -31,6 +31,7 @@
errors,
__version__ as _bzrlib_version,
)
+from bzrlib.transport.http import _urllib2_wrappers
# for testing, do
'''
@@ -53,6 +54,28 @@
errors.BzrError.__init__(self, url=url)
+class Transport(xmlrpclib.Transport):
+
+ def __init__(self, scheme, use_datetime=0):
+ xmlrpclib.Transport.__init__(self, use_datetime=use_datetime)
+ self._scheme = scheme
+ self._opener = _urllib2_wrappers.Opener()
+ self.verbose = 0
+
+ def request(self, host, handler, request_body, verbose=0):
+ self.verbose = verbose
+ url = self._scheme + "://" + host + handler
+ request = _urllib2_wrappers.Request("POST", url, request_body)
+ request.add_header("User-Agent", self.user_agent)
+ request.add_header("Content-Type", "text/xml")
+
+ response = self._opener.open(request)
+ if response.code != 200:
+ raise xmlrpclib.ProtocolError(host + handler, response.code,
+ response.msg, response.info())
+ return self.parse_response(response)
+
+
class LaunchpadService(object):
"""A service to talk to Launchpad via XMLRPC.
@@ -90,10 +113,7 @@
self._lp_instance = lp_instance
if transport is None:
uri_type = urllib.splittype(self.service_url)[0]
- if uri_type == 'https':
- transport = xmlrpclib.SafeTransport()
- else:
- transport = xmlrpclib.Transport()
+ transport = Transport(uri_type)
transport.user_agent = 'bzr/%s (xmlrpclib/%s)' \
% (_bzrlib_version, xmlrpclib.__version__)
self.transport = transport
=== modified file 'bzrlib/plugins/launchpad/test_lp_directory.py'
--- a/bzrlib/plugins/launchpad/test_lp_directory.py 2009-03-23 14:59:43 +0000
+++ b/bzrlib/plugins/launchpad/test_lp_directory.py 2009-10-30 14:44:52 +0000
@@ -16,10 +16,12 @@
"""Tests for directory lookup through Launchpad.net"""
+import os
import xmlrpclib
from bzrlib import (
errors,
+ tests,
)
from bzrlib.branch import Branch
from bzrlib.directory_service import directories
@@ -28,10 +30,38 @@
TestCaseWithMemoryTransport
)
from bzrlib.transport import get_transport
-from bzrlib.plugins.launchpad import _register_directory
+from bzrlib.plugins.launchpad import (
+ _register_directory,
+ lp_registration,
+ )
from bzrlib.plugins.launchpad.lp_directory import (
LaunchpadDirectory)
from bzrlib.plugins.launchpad.account import get_lp_login
+from bzrlib.tests import (
+ http_server,
+ http_utils,
+ )
+
+
+def load_tests(standard_tests, module, loader):
+ result = loader.suiteClass()
+ t_tests, remaining_tests = tests.split_suite_by_condition(
+ standard_tests, tests.condition_isinstance((
+ TestTransport,
+ )))
+ transport_scenarios = [
+ ('http', dict(server_class=PreCannedHTTPServer,)),
+ ]
+ if tests.HTTPSServerFeature.available():
+ transport_scenarios.append(
+ ('https', dict(server_class=PreCannedHTTPSServer,)),
+ )
+ tests.multiply_tests(t_tests, transport_scenarios, result)
+
+ # No parametrization for the remaining tests
+ result.addTests(remaining_tests)
+
+ return result
class FakeResolveFactory(object):
@@ -190,3 +220,108 @@
transport = get_transport('lp:///apt')
branch = Branch.open_from_transport(transport)
self.assertEqual(target_branch.base, branch.base)
+
+
+class PredefinedRequestHandler(http_server.TestingHTTPRequestHandler):
+ """Request handler for a unique and pre-defined request.
+
+ The only thing we care about here is that we receive a connection. But
+ since we want to dialog with a real http client, we have to send it correct
+ responses.
+
+ We expect to receive a *single* request nothing more (and we won't even
+ check what request it is), the tests will recognize us from our response.
+ """
+
+ def handle_one_request(self):
+ tcs = self.server.test_case_server
+ requestline = self.rfile.readline()
+ headers = self.MessageClass(self.rfile, 0)
+ if requestline.startswith('POST'):
+ # The body should be a single line (or we don't know where it ends
+ # and we don't want to issue a blocking read)
+ body = self.rfile.readline()
+
+ self.wfile.write(tcs.canned_response)
+
+class PreCannedServerMixin(object):
+
+ def __init__(self):
+ super(PreCannedServerMixin, self).__init__(
+ request_handler=PredefinedRequestHandler)
+ # Bytes read and written by the server
+ self.bytes_read = 0
+ self.bytes_written = 0
+ self.canned_response = None
+
+
+class PreCannedHTTPServer(PreCannedServerMixin, http_server.HttpServer):
+ pass
+
+
+if tests.HTTPSServerFeature.available():
+ from bzrlib.tests import https_server
+ class PreCannedHTTPSServer(PreCannedServerMixin, https_server.HTTPSServer):
+ pass
+
+
+class TestTransport(tests.TestCase):
+
+ # set by load_tests
+ server_class = None
+
+ def setUp(self):
+ tests.TestCase.setUp(self)
+ self.server = self.server_class()
+ self.server.setUp()
+ # Ensure we don't clobber env
+ self._captureVar('BZR_LP_XMLRPC_URL', None)
+
+ def tearDown(self):
+ self.server.tearDown()
+ tests.TestCase.tearDown(self)
+
+ def set_canned_response(self, server, path):
+ response_format = '''HTTP/1.1 200 OK\r
+Date: Tue, 11 Jul 2006 04:32:56 GMT\r
+Server: Apache/2.0.54 (Fedora)\r
+Last-Modified: Sun, 23 Apr 2006 19:35:20 GMT\r
+ETag: "56691-23-38e9ae00"\r
+Accept-Ranges: bytes\r
+Content-Length: %(length)d\r
+Connection: close\r
+Content-Type: text/plain; charset=UTF-8\r
+\r
+<?xml version='1.0'?>
+<methodResponse>
+<params>
+<param>
+<value><struct>
+<member>
+<name>urls</name>
+<value><array><data>
+<value><string>bzr+ssh://bazaar.launchpad.net/%(path)s</string></value>
+<value><string>http://bazaar.launchpad.net/%(path)s</string></value>
+</data></array></value>
+</member>
+</struct></value>
+</param>
+</params>
+</methodResponse>
+'''
+ length = 334 + 2 * len(path)
+ server.canned_response = response_format % dict(length=length,
+ path=path)
+
+ def test_simple_request(self):
+ self.set_canned_response(self.server, '~bzr-pqm/bzr/bzr.dev')
+ os.environ['BZR_LP_XMLRPC_URL'] = self.server.get_url()
+ service = lp_registration.LaunchpadService()
+ resolve = lp_registration.ResolveLaunchpadPathRequest('bzr')
+ result = resolve.submit(service)
+ urls = result.get('urls', None)
+ self.assertIsNot(None, urls)
+ self.assertEquals(
+ ['bzr+ssh://bazaar.launchpad.net/~bzr-pqm/bzr/bzr.dev',
+ 'http://bazaar.launchpad.net/~bzr-pqm/bzr/bzr.dev'],
+ urls)
More information about the bazaar-commits
mailing list