← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_smart_transport.py

  • Committer: Vincent Ladeuil
  • Date: 2008-01-03 08:49:38 UTC
  • mfrom: (3111.1.31 175524)
  • mto: This revision was merged to the branch mainline in revision 3158.
  • Revision ID: v.ladeuil+lp@free.fr-20080103084938-7kvurk5uvde2ui54
https://launchpad.net/bugs/175524
Fix bug #175524, http test servers are 1.1 compliant

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_smart_transport.py
39
39
        server,
40
40
        vfs,
41
41
)
42
 
from bzrlib.tests.HTTPTestUtil import (
 
42
from bzrlib.tests.http_utils import (
43
43
        HTTPServerWithSmarts,
44
44
        SmartRequestHandler,
45
45
        )
2322
2322
        return None
2323
2323
 
2324
2324
 
2325
 
class HTTPTunnellingSmokeTest(tests.TestCaseWithTransport):
2326
 
    
 
2325
class HTTPTunnellingSmokeTest(tests.TestCase):
 
2326
 
2327
2327
    def setUp(self):
2328
2328
        super(HTTPTunnellingSmokeTest, self).setUp()
2329
2329
        # We use the VFS layer as part of HTTP tunnelling tests.
2330
2330
        self._captureVar('BZR_NO_SMART_VFS', None)
2331
2331
 
2332
 
    def _test_bulk_data(self, url_protocol):
2333
 
        # We should be able to send and receive bulk data in a single message.
2334
 
        # The 'readv' command in the smart protocol both sends and receives bulk
2335
 
        # data, so we use that.
2336
 
        self.build_tree(['data-file'])
2337
 
        self.transport_readonly_server = HTTPServerWithSmarts
2338
 
 
2339
 
        http_transport = self.get_readonly_transport()
2340
 
        medium = http_transport.get_smart_medium()
2341
 
        # Since we provide the medium, the url below will be mostly ignored
2342
 
        # during the test, as long as the path is '/'.
2343
 
        remote_transport = remote.RemoteTransport('bzr://fake_host/',
2344
 
                                                  medium=medium)
2345
 
        self.assertEqual(
2346
 
            [(0, "c")], list(remote_transport.readv("data-file", [(0,1)])))
2347
 
 
2348
 
    def test_bulk_data_pycurl(self):
2349
 
        try:
2350
 
            self._test_bulk_data('http+pycurl')
2351
 
        except errors.UnsupportedProtocol, e:
2352
 
            raise tests.TestSkipped(str(e))
2353
 
    
2354
 
    def test_bulk_data_urllib(self):
2355
 
        self._test_bulk_data('http+urllib')
2356
 
 
2357
2332
    def test_smart_http_medium_request_accept_bytes(self):
2358
2333
        medium = FakeHTTPMedium()
2359
2334
        request = SmartClientHTTPMediumRequest(medium)
2363
2338
        request.finished_writing()
2364
2339
        self.assertEqual('abcdef', medium.written_request)
2365
2340
 
2366
 
    def _test_http_send_smart_request(self, url_protocol):
2367
 
        http_server = HTTPServerWithSmarts()
2368
 
        http_server._url_protocol = url_protocol
2369
 
        http_server.setUp(self.get_vfs_only_server())
2370
 
        self.addCleanup(http_server.tearDown)
2371
 
 
2372
 
        post_body = 'hello\n'
2373
 
        expected_reply_body = 'ok\x012\n'
2374
 
 
2375
 
        http_transport = get_transport(http_server.get_url())
2376
 
        medium = http_transport.get_smart_medium()
2377
 
        response = medium.send_http_smart_request(post_body)
2378
 
        reply_body = response.read()
2379
 
        self.assertEqual(expected_reply_body, reply_body)
2380
 
 
2381
 
    def test_http_send_smart_request_pycurl(self):
2382
 
        try:
2383
 
            self._test_http_send_smart_request('http+pycurl')
2384
 
        except errors.UnsupportedProtocol, e:
2385
 
            raise tests.TestSkipped(str(e))
2386
 
 
2387
 
    def test_http_send_smart_request_urllib(self):
2388
 
        self._test_http_send_smart_request('http+urllib')
2389
 
 
2390
 
    def test_http_server_with_smarts(self):
2391
 
        self.transport_readonly_server = HTTPServerWithSmarts
2392
 
 
2393
 
        post_body = 'hello\n'
2394
 
        expected_reply_body = 'ok\x012\n'
2395
 
 
2396
 
        smart_server_url = self.get_readonly_url('.bzr/smart')
2397
 
        reply = urllib2.urlopen(smart_server_url, post_body).read()
2398
 
 
2399
 
        self.assertEqual(expected_reply_body, reply)
2400
 
 
2401
 
    def test_smart_http_server_post_request_handler(self):
2402
 
        self.transport_readonly_server = HTTPServerWithSmarts
2403
 
        httpd = self.get_readonly_server()._get_httpd()
2404
 
 
2405
 
        socket = SampleSocket(
2406
 
            'POST /.bzr/smart HTTP/1.0\r\n'
2407
 
            # HTTP/1.0 posts must have a Content-Length.
2408
 
            'Content-Length: 6\r\n'
2409
 
            '\r\n'
2410
 
            'hello\n')
2411
 
        # Beware: the ('localhost', 80) below is the
2412
 
        # client_address parameter, but we don't have one because
2413
 
        # we have defined a socket which is not bound to an
2414
 
        # address. The test framework never uses this client
2415
 
        # address, so far...
2416
 
        request_handler = SmartRequestHandler(socket, ('localhost', 80), httpd)
2417
 
        response = socket.writefile.getvalue()
2418
 
        self.assertStartsWith(response, 'HTTP/1.0 200 ')
2419
 
        # This includes the end of the HTTP headers, and all the body.
2420
 
        expected_end_of_response = '\r\n\r\nok\x012\n'
2421
 
        self.assertEndsWith(response, expected_end_of_response)
2422
 
 
2423
 
 
2424
 
class SampleSocket(object):
2425
 
    """A socket-like object for use in testing the HTTP request handler."""
2426
 
    
2427
 
    def __init__(self, socket_read_content):
2428
 
        """Constructs a sample socket.
2429
 
 
2430
 
        :param socket_read_content: a byte sequence
2431
 
        """
2432
 
        # Use plain python StringIO so we can monkey-patch the close method to
2433
 
        # not discard the contents.
2434
 
        from StringIO import StringIO
2435
 
        self.readfile = StringIO(socket_read_content)
2436
 
        self.writefile = StringIO()
2437
 
        self.writefile.close = lambda: None
2438
 
        
2439
 
    def makefile(self, mode='r', bufsize=None):
2440
 
        if 'r' in mode:
2441
 
            return self.readfile
2442
 
        else:
2443
 
            return self.writefile
2444
 
 
2445
2341
 
2446
2342
class RemoteHTTPTransportTestCase(tests.TestCase):
2447
2343