← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_http.py

  • Committer: John Arbash Meinel
  • Date: 2006-11-06 21:00:45 UTC
  • mto: This revision was merged to the branch mainline in revision 2122.
  • Revision ID: john@arbash-meinel.com-20061106210045-b88ae833b5443145
Remove tabs from source files, and add a test to keep it that way.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_http.py
 
1
# Copyright (C) 2005, 2006 Canonical Ltd
 
2
#
 
3
# This program is free software; you can redistribute it and/or modify
 
4
# it under the terms of the GNU General Public License as published by
 
5
# the Free Software Foundation; either version 2 of the License, or
 
6
# (at your option) any later version.
 
7
#
 
8
# This program is distributed in the hope that it will be useful,
 
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
11
# GNU General Public License for more details.
 
12
#
 
13
# You should have received a copy of the GNU General Public License
 
14
# along with this program; if not, write to the Free Software
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
16
 
 
17
# FIXME: This test should be repeated for each available http client
 
18
# implementation; at the moment we have urllib and pycurl.
 
19
 
 
20
# TODO: Should be renamed to bzrlib.transport.http.tests?
 
21
 
 
22
import errno
 
23
import select
 
24
import socket
 
25
import threading
 
26
 
 
27
import bzrlib
 
28
from bzrlib.errors import DependencyNotPresent, UnsupportedProtocol
 
29
from bzrlib import osutils
 
30
from bzrlib.tests import TestCase, TestSkipped
 
31
from bzrlib.transport import get_transport, Transport
 
32
from bzrlib.transport.http import extract_auth, HttpTransportBase
 
33
from bzrlib.transport.http._urllib import HttpTransport_urllib
 
34
from bzrlib.tests.HTTPTestUtil import TestCaseWithWebserver
 
35
 
 
36
 
 
37
class FakeManager(object):
 
38
 
 
39
    def __init__(self):
 
40
        self.credentials = []
 
41
        
 
42
    def add_password(self, realm, host, username, password):
 
43
        self.credentials.append([realm, host, username, password])
 
44
 
 
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
 
 
100
class TestHttpUrls(TestCase):
 
101
 
 
102
    def test_url_parsing(self):
 
103
        f = FakeManager()
 
104
        url = extract_auth('http://example.com', f)
 
105
        self.assertEquals('http://example.com', url)
 
106
        self.assertEquals(0, len(f.credentials))
 
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)
 
109
        self.assertEquals(1, len(f.credentials))
 
110
        self.assertEquals([None, 'www.bazaar-vcs.org', 'user', 'pass'], f.credentials[0])
 
111
        
 
112
    def test_abs_url(self):
 
113
        """Construction of absolute http URLs"""
 
114
        t = HttpTransport_urllib('http://bazaar-vcs.org/bzr/bzr.dev/')
 
115
        eq = self.assertEqualDiff
 
116
        eq(t.abspath('.'),
 
117
           'http://bazaar-vcs.org/bzr/bzr.dev')
 
118
        eq(t.abspath('foo/bar'), 
 
119
           'http://bazaar-vcs.org/bzr/bzr.dev/foo/bar')
 
120
        eq(t.abspath('.bzr'),
 
121
           'http://bazaar-vcs.org/bzr/bzr.dev/.bzr')
 
122
        eq(t.abspath('.bzr/1//2/./3'),
 
123
           'http://bazaar-vcs.org/bzr/bzr.dev/.bzr/1/2/3')
 
124
 
 
125
    def test_invalid_http_urls(self):
 
126
        """Trap invalid construction of urls"""
 
127
        t = HttpTransport_urllib('http://bazaar-vcs.org/bzr/bzr.dev/')
 
128
        self.assertRaises(ValueError,
 
129
            t.abspath,
 
130
            '.bzr/')
 
131
 
 
132
    def test_http_root_urls(self):
 
133
        """Construction of URLs from server root"""
 
134
        t = HttpTransport_urllib('http://bzr.ozlabs.org/')
 
135
        eq = self.assertEqualDiff
 
136
        eq(t.abspath('.bzr/tree-version'),
 
137
           'http://bzr.ozlabs.org/.bzr/tree-version')
 
138
 
 
139
    def test_http_impl_urls(self):
 
140
        """There are servers which ask for particular clients to connect"""
 
141
        try:
 
142
            from bzrlib.transport.http._pycurl import HttpServer_PyCurl
 
143
            server = HttpServer_PyCurl()
 
144
            try:
 
145
                server.setUp()
 
146
                url = server.get_url()
 
147
                self.assertTrue(url.startswith('http+pycurl://'))
 
148
            finally:
 
149
                server.tearDown()
 
150
        except DependencyNotPresent:
 
151
            raise TestSkipped('pycurl not present')
 
152
 
 
153
 
 
154
class TestHttpMixins(object):
 
155
 
 
156
    def _prep_tree(self):
 
157
        self.build_tree(['xxx', 'foo/', 'foo/bar'], line_endings='binary',
 
158
                        transport=self.get_transport())
 
159
 
 
160
    def test_http_has(self):
 
161
        server = self.get_readonly_server()
 
162
        t = self._transport(server.get_url())
 
163
        self.assertEqual(t.has('foo/bar'), True)
 
164
        self.assertEqual(len(server.logs), 1)
 
165
        self.assertContainsRe(server.logs[0], 
 
166
            r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "bzr/')
 
167
 
 
168
    def test_http_has_not_found(self):
 
169
        server = self.get_readonly_server()
 
170
        t = self._transport(server.get_url())
 
171
        self.assertEqual(t.has('not-found'), False)
 
172
        self.assertContainsRe(server.logs[1], 
 
173
            r'"HEAD /not-found HTTP/1.." 404 - "-" "bzr/')
 
174
 
 
175
    def test_http_get(self):
 
176
        server = self.get_readonly_server()
 
177
        t = self._transport(server.get_url())
 
178
        fp = t.get('foo/bar')
 
179
        self.assertEqualDiff(
 
180
            fp.read(),
 
181
            'contents of foo/bar\n')
 
182
        self.assertEqual(len(server.logs), 1)
 
183
        self.assertTrue(server.logs[0].find(
 
184
            '"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s' % bzrlib.__version__) > -1)
 
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
        
 
193
 
 
194
class TestHttpConnections_urllib(TestCaseWithWebserver, TestHttpMixins):
 
195
 
 
196
    _transport = HttpTransport_urllib
 
197
 
 
198
    def setUp(self):
 
199
        TestCaseWithWebserver.setUp(self)
 
200
        self._prep_tree()
 
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')
 
210
 
 
211
 
 
212
class TestHttpConnections_pycurl(TestCaseWithWebserver, TestHttpMixins):
 
213
 
 
214
    def _get_pycurl_maybe(self):
 
215
        try:
 
216
            from bzrlib.transport.http._pycurl import PyCurlTransport
 
217
            return PyCurlTransport
 
218
        except DependencyNotPresent:
 
219
            raise TestSkipped('pycurl not present')
 
220
 
 
221
    _transport = property(_get_pycurl_maybe)
 
222
 
 
223
    def setUp(self):
 
224
        TestCaseWithWebserver.setUp(self)
 
225
        self._prep_tree()
 
226
 
 
227
 
 
228
class TestHttpTransportRegistration(TestCase):
 
229
    """Test registrations of various http implementations"""
 
230
 
 
231
    def test_http_registered(self):
 
232
        import bzrlib.transport.http._urllib
 
233
        from bzrlib.transport import get_transport
 
234
        # urlllib should always be present
 
235
        t = get_transport('http+urllib://bzr.google.com/')
 
236
        self.assertIsInstance(t, Transport)
 
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)