← 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-08-14 16:16:53 UTC
  • mto: (1946.2.6 reduce-knit-churn)
  • mto: This revision was merged to the branch mainline in revision 1919.
  • Revision ID: john@arbash-meinel.com-20060814161653-54cdcdadcd4e9003
Remove bogus entry from BRANCH.TODO

Show diffs side-by-side

added added

removed removed

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