/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-15 16:22:42 UTC
  • mfrom: (2077 +trunk)
  • mto: This revision was merged to the branch mainline in revision 2078.
  • Revision ID: aaron.bentley@utoronto.ca-20061015162242-2de5677dd0495500
Merge from bzr.dev

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_http.py
19
19
 
20
20
# TODO: Should be renamed to bzrlib.transport.http.tests?
21
21
 
 
22
import errno
 
23
import select
22
24
import socket
 
25
import threading
23
26
 
24
27
import bzrlib
25
 
from bzrlib.errors import DependencyNotPresent
 
28
from bzrlib.errors import DependencyNotPresent, UnsupportedProtocol
26
29
from bzrlib.tests import TestCase, TestSkipped
27
 
from bzrlib.transport import Transport
 
30
from bzrlib.transport import get_transport, Transport
28
31
from bzrlib.transport.http import extract_auth, HttpTransportBase
29
32
from bzrlib.transport.http._urllib import HttpTransport_urllib
30
33
from bzrlib.tests.HTTPTestUtil import TestCaseWithWebserver
31
34
 
32
35
 
33
 
class FakeManager (object):
 
36
class FakeManager(object):
34
37
 
35
38
    def __init__(self):
36
39
        self.credentials = []
39
42
        self.credentials.append([realm, host, username, password])
40
43
 
41
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
 
42
99
class TestHttpUrls(TestCase):
43
100
 
44
101
    def test_url_parsing(self):
125
182
        self.assertTrue(server.logs[0].find(
126
183
            '"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s' % bzrlib.__version__) > -1)
127
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
        
128
192
 
129
193
class TestHttpConnections_urllib(TestCaseWithWebserver, TestHttpMixins):
130
194
 
160
224
        self._prep_tree()
161
225
 
162
226
 
163
 
 
164
227
class TestHttpTransportRegistration(TestCase):
165
228
    """Test registrations of various http implementations"""
166
229
 
197
260
        self.assertEqual([[10, 12], [22, 26]], ranges)
198
261
 
199
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
 
200
291
class TestRangeHeader(TestCase):
201
292
    """Test range_header method"""
202
293
 
220
311
        self.check_header('0-9,300-5000,-50',
221
312
                          ranges=[(0,9), (300,5000)],
222
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)