/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: Aaron Bentley
  • Date: 2006-10-20 01:46:22 UTC
  • mto: This revision was merged to the branch mainline in revision 2098.
  • Revision ID: aaron.bentley@utoronto.ca-20061020014622-745c98fdc87811a8
When 'directory' path element isn't a directory, return None from path2id

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