← 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: John Arbash Meinel
  • Date: 2010-02-17 17:11:16 UTC
  • mfrom: (4797.2.17 2.1)
  • mto: (4797.2.18 2.1)
  • mto: This revision was merged to the branch mainline in revision 5055.
  • Revision ID: john@arbash-meinel.com-20100217171116-h7t9223ystbnx5h8
merge bzr.2.1 in preparation for NEWS entry.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/plugins/launchpad/test_lp_directory.py
1
 
# Copyright (C) 2007, 2008 Canonical Ltd
 
1
# Copyright (C) 2007-2010 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
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.start_server()
 
278
        # Ensure we don't clobber env
 
279
        self._captureVar('BZR_LP_XMLRPC_URL', None)
 
280
 
 
281
    def tearDown(self):
 
282
        self.server.stop_server()
 
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