/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_http.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2006-11-16 18:33:33 UTC
  • mfrom: (2138.1.1 robuster.external.diff)
  • Revision ID: pqm@pqm.ubuntu.com-20061116183333-5f56523d7b44e564
(Dmitry Vasiliev) Robuster external diff output handling.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_http.py
1
 
# Copyright (C) 2005, 2006 Canonical
 
1
# Copyright (C) 2005, 2006 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
19
19
 
20
20
# TODO: Should be renamed to bzrlib.transport.http.tests?
21
21
 
 
22
import errno
 
23
import select
 
24
import socket
 
25
import threading
 
26
 
22
27
import bzrlib
23
 
from bzrlib.errors import DependencyNotPresent
 
28
from bzrlib.errors import DependencyNotPresent, UnsupportedProtocol
 
29
from bzrlib import osutils
24
30
from bzrlib.tests import TestCase, TestSkipped
25
 
from bzrlib.transport import Transport
26
 
from bzrlib.transport.http import extract_auth
 
31
from bzrlib.transport import get_transport, Transport
 
32
from bzrlib.transport.http import extract_auth, HttpTransportBase
27
33
from bzrlib.transport.http._urllib import HttpTransport_urllib
28
34
from bzrlib.tests.HTTPTestUtil import TestCaseWithWebserver
29
35
 
30
 
class FakeManager (object):
 
36
 
 
37
class FakeManager(object):
 
38
 
31
39
    def __init__(self):
32
40
        self.credentials = []
33
41
        
35
43
        self.credentials.append([realm, host, username, password])
36
44
 
37
45
 
 
46
class RecordingServer(object):
 
47
    """A fake HTTP server.
 
48
    
 
49
    It records the bytes sent to it, and replies with a 200.
 
50
    """
 
51
 
 
52
    def __init__(self, expect_body_tail=None):
 
53
        """Constructor.
 
54
 
 
55
        :type expect_body_tail: str
 
56
        :param expect_body_tail: a reply won't be sent until this string is
 
57
            received.
 
58
        """
 
59
        self._expect_body_tail = expect_body_tail
 
60
        self.host = None
 
61
        self.port = None
 
62
        self.received_bytes = ''
 
63
 
 
64
    def setUp(self):
 
65
        self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 
66
        self._sock.bind(('127.0.0.1', 0))
 
67
        self.host, self.port = self._sock.getsockname()
 
68
        self._ready = threading.Event()
 
69
        self._thread = threading.Thread(target=self._accept_read_and_reply)
 
70
        self._thread.setDaemon(True)
 
71
        self._thread.start()
 
72
        self._ready.wait(5)
 
73
 
 
74
    def _accept_read_and_reply(self):
 
75
        self._sock.listen(1)
 
76
        self._ready.set()
 
77
        self._sock.settimeout(5)
 
78
        try:
 
79
            conn, address = self._sock.accept()
 
80
            # On win32, the accepted connection will be non-blocking to start
 
81
            # with because we're using settimeout.
 
82
            conn.setblocking(True)
 
83
            while not self.received_bytes.endswith(self._expect_body_tail):
 
84
                self.received_bytes += conn.recv(4096)
 
85
            conn.sendall('HTTP/1.1 200 OK\r\n')
 
86
        except socket.timeout:
 
87
            # Make sure the client isn't stuck waiting for us to e.g. accept.
 
88
            self._sock.close()
 
89
 
 
90
    def tearDown(self):
 
91
        try:
 
92
            self._sock.close()
 
93
        except socket.error:
 
94
            # We might have already closed it.  We don't care.
 
95
            pass
 
96
        self.host = None
 
97
        self.port = None
 
98
 
 
99
 
38
100
class TestHttpUrls(TestCase):
 
101
 
39
102
    def test_url_parsing(self):
40
103
        f = FakeManager()
41
104
        url = extract_auth('http://example.com', f)
42
105
        self.assertEquals('http://example.com', url)
43
106
        self.assertEquals(0, len(f.credentials))
44
 
        url = extract_auth('http://user:pass@www.bazaar-ng.org/bzr/bzr.dev', f)
45
 
        self.assertEquals('http://www.bazaar-ng.org/bzr/bzr.dev', url)
 
107
        url = extract_auth('http://user:pass@www.bazaar-vcs.org/bzr/bzr.dev', f)
 
108
        self.assertEquals('http://www.bazaar-vcs.org/bzr/bzr.dev', url)
46
109
        self.assertEquals(1, len(f.credentials))
47
 
        self.assertEquals([None, 'www.bazaar-ng.org', 'user', 'pass'], f.credentials[0])
 
110
        self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'], f.credentials[0])
48
111
        
49
112
    def test_abs_url(self):
50
113
        """Construction of absolute http URLs"""
51
 
        t = HttpTransport_urllib('http://bazaar-ng.org/bzr/bzr.dev/')
 
114
        t = HttpTransport_urllib('http://bazaar-vcs.org/bzr/bzr.dev/')
52
115
        eq = self.assertEqualDiff
53
116
        eq(t.abspath('.'),
54
 
           'http://bazaar-ng.org/bzr/bzr.dev')
 
117
           'http://bazaar-vcs.org/bzr/bzr.dev')
55
118
        eq(t.abspath('foo/bar'), 
56
 
           'http://bazaar-ng.org/bzr/bzr.dev/foo/bar')
 
119
           'http://bazaar-vcs.org/bzr/bzr.dev/foo/bar')
57
120
        eq(t.abspath('.bzr'),
58
 
           'http://bazaar-ng.org/bzr/bzr.dev/.bzr')
 
121
           'http://bazaar-vcs.org/bzr/bzr.dev/.bzr')
59
122
        eq(t.abspath('.bzr/1//2/./3'),
60
 
           'http://bazaar-ng.org/bzr/bzr.dev/.bzr/1/2/3')
 
123
           'http://bazaar-vcs.org/bzr/bzr.dev/.bzr/1/2/3')
61
124
 
62
125
    def test_invalid_http_urls(self):
63
126
        """Trap invalid construction of urls"""
64
 
        t = HttpTransport_urllib('http://bazaar-ng.org/bzr/bzr.dev/')
 
127
        t = HttpTransport_urllib('http://bazaar-vcs.org/bzr/bzr.dev/')
65
128
        self.assertRaises(ValueError,
66
129
            t.abspath,
67
130
            '.bzr/')
68
 
        self.assertRaises(ValueError,
69
 
            t.abspath,
70
 
            '/.bzr')
71
131
 
72
132
    def test_http_root_urls(self):
73
133
        """Construction of URLs from server root"""
90
150
        except DependencyNotPresent:
91
151
            raise TestSkipped('pycurl not present')
92
152
 
 
153
 
93
154
class TestHttpMixins(object):
94
155
 
95
156
    def _prep_tree(self):
122
183
        self.assertTrue(server.logs[0].find(
123
184
            '"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s' % bzrlib.__version__) > -1)
124
185
 
 
186
    def test_get_smart_medium(self):
 
187
        # For HTTP, get_smart_medium should return the transport object.
 
188
        server = self.get_readonly_server()
 
189
        http_transport = self._transport(server.get_url())
 
190
        medium = http_transport.get_smart_medium()
 
191
        self.assertIs(medium, http_transport)
 
192
        
125
193
 
126
194
class TestHttpConnections_urllib(TestCaseWithWebserver, TestHttpMixins):
 
195
 
127
196
    _transport = HttpTransport_urllib
128
197
 
129
198
    def setUp(self):
130
199
        TestCaseWithWebserver.setUp(self)
131
200
        self._prep_tree()
132
201
 
 
202
    def test_has_on_bogus_host(self):
 
203
        import urllib2
 
204
        # Get a random address, so that we can be sure there is no
 
205
        # http handler there.
 
206
        s = socket.socket()
 
207
        s.bind(('localhost', 0))
 
208
        t = self._transport('http://%s:%s/' % s.getsockname())
 
209
        self.assertRaises(urllib2.URLError, t.has, 'foo/bar')
133
210
 
134
211
 
135
212
class TestHttpConnections_pycurl(TestCaseWithWebserver, TestHttpMixins):
148
225
        self._prep_tree()
149
226
 
150
227
 
151
 
 
152
228
class TestHttpTransportRegistration(TestCase):
153
229
    """Test registrations of various http implementations"""
154
230
 
159
235
        t = get_transport('http+urllib://bzr.google.com/')
160
236
        self.assertIsInstance(t, Transport)
161
237
        self.assertIsInstance(t, bzrlib.transport.http._urllib.HttpTransport_urllib)
 
238
 
 
239
 
 
240
class TestOffsets(TestCase):
 
241
    """Test offsets_to_ranges method"""
 
242
 
 
243
    def test_offsets_to_ranges_simple(self):
 
244
        to_range = HttpTransportBase.offsets_to_ranges
 
245
        ranges = to_range([(10, 1)])
 
246
        self.assertEqual([[10, 10]], ranges)
 
247
 
 
248
        ranges = to_range([(0, 1), (1, 1)])
 
249
        self.assertEqual([[0, 1]], ranges)
 
250
 
 
251
        ranges = to_range([(1, 1), (0, 1)])
 
252
        self.assertEqual([[0, 1]], ranges)
 
253
 
 
254
    def test_offset_to_ranges_overlapped(self):
 
255
        to_range = HttpTransportBase.offsets_to_ranges
 
256
 
 
257
        ranges = to_range([(10, 1), (20, 2), (22, 5)])
 
258
        self.assertEqual([[10, 10], [20, 26]], ranges)
 
259
 
 
260
        ranges = to_range([(10, 1), (11, 2), (22, 5)])
 
261
        self.assertEqual([[10, 12], [22, 26]], ranges)
 
262
 
 
263
 
 
264
class TestPost(TestCase):
 
265
 
 
266
    def _test_post_body_is_received(self, scheme):
 
267
        server = RecordingServer(expect_body_tail='end-of-body')
 
268
        server.setUp()
 
269
        self.addCleanup(server.tearDown)
 
270
        url = '%s://%s:%s/' % (scheme, server.host, server.port)
 
271
        try:
 
272
            http_transport = get_transport(url)
 
273
        except UnsupportedProtocol:
 
274
            raise TestSkipped('%s not available' % scheme)
 
275
        code, response = http_transport._post('abc def end-of-body')
 
276
        self.assertTrue(
 
277
            server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
 
278
        self.assertTrue('content-length: 19\r' in server.received_bytes.lower())
 
279
        # The transport should not be assuming that the server can accept
 
280
        # chunked encoding the first time it connects, because HTTP/1.1, so we
 
281
        # check for the literal string.
 
282
        self.assertTrue(
 
283
            server.received_bytes.endswith('\r\n\r\nabc def end-of-body'))
 
284
 
 
285
    def test_post_body_is_received_urllib(self):
 
286
        self._test_post_body_is_received('http+urllib')
 
287
 
 
288
    def test_post_body_is_received_pycurl(self):
 
289
        self._test_post_body_is_received('http+pycurl')
 
290
 
 
291
 
 
292
class TestRangeHeader(TestCase):
 
293
    """Test range_header method"""
 
294
 
 
295
    def check_header(self, value, ranges=[], tail=0):
 
296
        range_header = HttpTransportBase.range_header
 
297
        self.assertEqual(value, range_header(ranges, tail))
 
298
 
 
299
    def test_range_header_single(self):
 
300
        self.check_header('0-9', ranges=[[0,9]])
 
301
        self.check_header('100-109', ranges=[[100,109]])
 
302
 
 
303
    def test_range_header_tail(self):
 
304
        self.check_header('-10', tail=10)
 
305
        self.check_header('-50', tail=50)
 
306
 
 
307
    def test_range_header_multi(self):
 
308
        self.check_header('0-9,100-200,300-5000',
 
309
                          ranges=[(0,9), (100, 200), (300,5000)])
 
310
 
 
311
    def test_range_header_mixed(self):
 
312
        self.check_header('0-9,300-5000,-50',
 
313
                          ranges=[(0,9), (300,5000)],
 
314
                          tail=50)
 
315
 
 
316
        
 
317
class TestRecordingServer(TestCase):
 
318
 
 
319
    def test_create(self):
 
320
        server = RecordingServer(expect_body_tail=None)
 
321
        self.assertEqual('', server.received_bytes)
 
322
        self.assertEqual(None, server.host)
 
323
        self.assertEqual(None, server.port)
 
324
 
 
325
    def test_setUp_and_tearDown(self):
 
326
        server = RecordingServer(expect_body_tail=None)
 
327
        server.setUp()
 
328
        try:
 
329
            self.assertNotEqual(None, server.host)
 
330
            self.assertNotEqual(None, server.port)
 
331
        finally:
 
332
            server.tearDown()
 
333
        self.assertEqual(None, server.host)
 
334
        self.assertEqual(None, server.port)
 
335
 
 
336
    def test_send_receive_bytes(self):
 
337
        server = RecordingServer(expect_body_tail='c')
 
338
        server.setUp()
 
339
        self.addCleanup(server.tearDown)
 
340
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 
341
        sock.connect((server.host, server.port))
 
342
        sock.sendall('abc')
 
343
        self.assertEqual('HTTP/1.1 200 OK\r\n',
 
344
                         osutils.recv_all(sock, 4096))
 
345
        self.assertEqual('abc', server.received_bytes)