/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: Martin Pool
  • Date: 2006-10-19 02:59:24 UTC
  • mfrom: (2089 +trunk)
  • mto: This revision was merged to the branch mainline in revision 2090.
  • Revision ID: mbp@sourcefrog.net-20061019025924-1af1ed4510b4e71f
merge up bzr.dev

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
24
29
from bzrlib.tests import TestCase, TestSkipped
25
 
from bzrlib.transport import Transport
 
30
from bzrlib.transport import get_transport, Transport
26
31
from bzrlib.transport.http import extract_auth, HttpTransportBase
27
32
from bzrlib.transport.http._urllib import HttpTransport_urllib
28
33
from bzrlib.tests.HTTPTestUtil import TestCaseWithWebserver
29
34
 
30
35
 
31
 
class FakeManager (object):
 
36
class FakeManager(object):
32
37
 
33
38
    def __init__(self):
34
39
        self.credentials = []
37
42
        self.credentials.append([realm, host, username, password])
38
43
 
39
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
 
40
99
class TestHttpUrls(TestCase):
41
100
 
42
101
    def test_url_parsing(self):
68
127
        self.assertRaises(ValueError,
69
128
            t.abspath,
70
129
            '.bzr/')
71
 
        self.assertRaises(ValueError,
72
 
            t.abspath,
73
 
            '/.bzr')
74
130
 
75
131
    def test_http_root_urls(self):
76
132
        """Construction of URLs from server root"""
126
182
        self.assertTrue(server.logs[0].find(
127
183
            '"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s' % bzrlib.__version__) > -1)
128
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
        
129
192
 
130
193
class TestHttpConnections_urllib(TestCaseWithWebserver, TestHttpMixins):
131
194
 
135
198
        TestCaseWithWebserver.setUp(self)
136
199
        self._prep_tree()
137
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')
138
209
 
139
210
 
140
211
class TestHttpConnections_pycurl(TestCaseWithWebserver, TestHttpMixins):
153
224
        self._prep_tree()
154
225
 
155
226
 
156
 
 
157
227
class TestHttpTransportRegistration(TestCase):
158
228
    """Test registrations of various http implementations"""
159
229
 
190
260
        self.assertEqual([[10, 12], [22, 26]], ranges)
191
261
 
192
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
 
193
291
class TestRangeHeader(TestCase):
194
292
    """Test range_header method"""
195
293
 
213
311
        self.check_header('0-9,300-5000,-50',
214
312
                          ranges=[(0,9), (300,5000)],
215
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)