← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

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

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2009-11-03 03:58:26 UTC
  • mfrom: (4668.1.5 2.1.0b2)
  • Revision ID: pqm@pqm.ubuntu.com-20091103035826-tr4qa6fznzmirgiq
(jam) Merge 2.1.0b2 (-final) to bzr.dev

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/plugins/launchpad/test_lp_directory.py
16
16
 
17
17
"""Tests for directory lookup through Launchpad.net"""
18
18
 
 
19
import os
19
20
import xmlrpclib
20
21
 
21
22
from bzrlib import (
22
23
    errors,
 
24
    tests,
23
25
    )
24
26
from bzrlib.branch import Branch
25
27
from bzrlib.directory_service import directories
28
30
    TestCaseWithMemoryTransport
29
31
)
30
32
from bzrlib.transport import get_transport
31
 
from bzrlib.plugins.launchpad import _register_directory
 
33
from bzrlib.plugins.launchpad import (
 
34
    _register_directory,
 
35
    lp_registration,
 
36
    )
32
37
from bzrlib.plugins.launchpad.lp_directory import (
33
38
    LaunchpadDirectory)
34
39
from bzrlib.plugins.launchpad.account import get_lp_login
 
40
from bzrlib.tests import (
 
41
    http_server,
 
42
    http_utils,
 
43
    )
 
44
 
 
45
 
 
46
def load_tests(standard_tests, module, loader):
 
47
    result = loader.suiteClass()
 
48
    t_tests, remaining_tests = tests.split_suite_by_condition(
 
49
        standard_tests, tests.condition_isinstance((
 
50
                TestXMLRPCTransport,
 
51
                )))
 
52
    transport_scenarios = [
 
53
        ('http', dict(server_class=PreCannedHTTPServer,)),
 
54
        ]
 
55
    if tests.HTTPSServerFeature.available():
 
56
        transport_scenarios.append(
 
57
            ('https', dict(server_class=PreCannedHTTPSServer,)),
 
58
            )
 
59
    tests.multiply_tests(t_tests, transport_scenarios, result)
 
60
 
 
61
    # No parametrization for the remaining tests
 
62
    result.addTests(remaining_tests)
 
63
 
 
64
    return result
35
65
 
36
66
 
37
67
class FakeResolveFactory(object):
190
220
        transport = get_transport('lp:///apt')
191
221
        branch = Branch.open_from_transport(transport)
192
222
        self.assertEqual(target_branch.base, branch.base)
 
223
 
 
224
 
 
225
class PredefinedRequestHandler(http_server.TestingHTTPRequestHandler):
 
226
    """Request handler for a unique and pre-defined request.
 
227
 
 
228
    The only thing we care about here is that we receive a connection. But
 
229
    since we want to dialog with a real http client, we have to send it correct
 
230
    responses.
 
231
 
 
232
    We expect to receive a *single* request nothing more (and we won't even
 
233
    check what request it is), the tests will recognize us from our response.
 
234
    """
 
235
 
 
236
    def handle_one_request(self):
 
237
        tcs = self.server.test_case_server
 
238
        requestline = self.rfile.readline()
 
239
        headers = self.MessageClass(self.rfile, 0)
 
240
        if requestline.startswith('POST'):
 
241
            # The body should be a single line (or we don't know where it ends
 
242
            # and we don't want to issue a blocking read)
 
243
            body = self.rfile.readline()
 
244
 
 
245
        self.wfile.write(tcs.canned_response)
 
246
 
 
247
 
 
248
class PreCannedServerMixin(object):
 
249
 
 
250
    def __init__(self):
 
251
        super(PreCannedServerMixin, self).__init__(
 
252
            request_handler=PredefinedRequestHandler)
 
253
        # Bytes read and written by the server
 
254
        self.bytes_read = 0
 
255
        self.bytes_written = 0
 
256
        self.canned_response = None
 
257
 
 
258
 
 
259
class PreCannedHTTPServer(PreCannedServerMixin, http_server.HttpServer):
 
260
    pass
 
261
 
 
262
 
 
263
if tests.HTTPSServerFeature.available():
 
264
    from bzrlib.tests import https_server
 
265
    class PreCannedHTTPSServer(PreCannedServerMixin, https_server.HTTPSServer):
 
266
        pass
 
267
 
 
268
 
 
269
class TestXMLRPCTransport(tests.TestCase):
 
270
 
 
271
    # set by load_tests
 
272
    server_class = None
 
273
 
 
274
    def setUp(self):
 
275
        tests.TestCase.setUp(self)
 
276
        self.server = self.server_class()
 
277
        self.server.setUp()
 
278
        # Ensure we don't clobber env
 
279
        self._captureVar('BZR_LP_XMLRPC_URL', None)
 
280
 
 
281
    def tearDown(self):
 
282
        self.server.tearDown()
 
283
        tests.TestCase.tearDown(self)
 
284
 
 
285
    def set_canned_response(self, server, path):
 
286
        response_format = '''HTTP/1.1 200 OK\r
 
287
Date: Tue, 11 Jul 2006 04:32:56 GMT\r
 
288
Server: Apache/2.0.54 (Fedora)\r
 
289
Last-Modified: Sun, 23 Apr 2006 19:35:20 GMT\r
 
290
ETag: "56691-23-38e9ae00"\r
 
291
Accept-Ranges: bytes\r
 
292
Content-Length: %(length)d\r
 
293
Connection: close\r
 
294
Content-Type: text/plain; charset=UTF-8\r
 
295
\r
 
296
<?xml version='1.0'?>
 
297
<methodResponse>
 
298
<params>
 
299
<param>
 
300
<value><struct>
 
301
<member>
 
302
<name>urls</name>
 
303
<value><array><data>
 
304
<value><string>bzr+ssh://bazaar.launchpad.net/%(path)s</string></value>
 
305
<value><string>http://bazaar.launchpad.net/%(path)s</string></value>
 
306
</data></array></value>
 
307
</member>
 
308
</struct></value>
 
309
</param>
 
310
</params>
 
311
</methodResponse>
 
312
'''
 
313
        length = 334 + 2 * len(path)
 
314
        server.canned_response = response_format % dict(length=length,
 
315
                                                        path=path)
 
316
 
 
317
    def do_request(self, server_url):
 
318
        os.environ['BZR_LP_XMLRPC_URL'] = self.server.get_url()
 
319
        service = lp_registration.LaunchpadService()
 
320
        resolve = lp_registration.ResolveLaunchpadPathRequest('bzr')
 
321
        result = resolve.submit(service)
 
322
        return result
 
323
 
 
324
    def test_direct_request(self):
 
325
        self.set_canned_response(self.server, '~bzr-pqm/bzr/bzr.dev')
 
326
        result = self.do_request(self.server.get_url())
 
327
        urls = result.get('urls', None)
 
328
        self.assertIsNot(None, urls)
 
329
        self.assertEquals(
 
330
            ['bzr+ssh://bazaar.launchpad.net/~bzr-pqm/bzr/bzr.dev',
 
331
             'http://bazaar.launchpad.net/~bzr-pqm/bzr/bzr.dev'],
 
332
            urls)
 
333
    # FIXME: we need to test with a real proxy, I can't find a way so simulate
 
334
    # CONNECT without leaving one server hanging the test :-/ Since that maybe
 
335
    # related to the leaking tests problems, I'll punt for now -- vila 20091030