← 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: Martin Pool
  • Date: 2006-11-02 10:20:19 UTC
  • mfrom: (2114 +trunk)
  • mto: This revision was merged to the branch mainline in revision 2119.
  • Revision ID: mbp@sourcefrog.net-20061102102019-9a5a02f485dff6f6
merge bzr.dev and reconcile several changes, also some test fixes

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
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
 
29
from bzrlib import osutils
26
30
from bzrlib.tests import TestCase, TestSkipped
27
 
from bzrlib.transport import Transport
 
31
from bzrlib.transport import get_transport, Transport
28
32
from bzrlib.transport.http import extract_auth, HttpTransportBase
29
33
from bzrlib.transport.http._urllib import HttpTransport_urllib
30
34
from bzrlib.tests.HTTPTestUtil import TestCaseWithWebserver
31
35
 
32
36
 
33
 
class FakeManager (object):
 
37
class FakeManager(object):
34
38
 
35
39
    def __init__(self):
36
40
        self.credentials = []
39
43
        self.credentials.append([realm, host, username, password])
40
44
 
41
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
 
42
100
class TestHttpUrls(TestCase):
43
101
 
44
102
    def test_url_parsing(self):
125
183
        self.assertTrue(server.logs[0].find(
126
184
            '"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s' % bzrlib.__version__) > -1)
127
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
        
128
193
 
129
194
class TestHttpConnections_urllib(TestCaseWithWebserver, TestHttpMixins):
130
195
 
160
225
        self._prep_tree()
161
226
 
162
227
 
163
 
 
164
228
class TestHttpTransportRegistration(TestCase):
165
229
    """Test registrations of various http implementations"""
166
230
 
197
261
        self.assertEqual([[10, 12], [22, 26]], ranges)
198
262
 
199
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
 
200
292
class TestRangeHeader(TestCase):
201
293
    """Test range_header method"""
202
294
 
220
312
        self.check_header('0-9,300-5000,-50',
221
313
                          ranges=[(0,9), (300,5000)],
222
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)