← 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: Vincent Ladeuil
  • Date: 2007-12-19 09:29:21 UTC
  • mto: (3146.3.1 179368) (3156.2.1 trunk)
  • mto: This revision was merged to the branch mainline in revision 3158.
  • Revision ID: v.ladeuil+lp@free.fr-20071219092921-3oxvnp5ptfhz8v42
Further refactoring.

* bzrlib/tests/test_http_implementations.py:
(TestHttpConnections, TestPost): Tranferred from test_http.py (use
case for moved lines).

* bzrlib/tests/http_utils.py
(RecordingServer): Transfered from test_http.py (use case for
moved lines as are the surrounding commits related to test_http.py
and test_http_implementations.py).

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_http.py
39
39
    )
40
40
from bzrlib.tests import (
41
41
    http_server,
 
42
    http_utils,
42
43
    TestCase,
43
44
    TestUIFactory,
44
45
    TestSkipped,
91
92
        self.credentials.append([realm, host, username, password])
92
93
 
93
94
 
94
 
class RecordingServer(object):
95
 
    """A fake HTTP server.
96
 
    
97
 
    It records the bytes sent to it, and replies with a 200.
98
 
    """
99
 
 
100
 
    def __init__(self, expect_body_tail=None):
101
 
        """Constructor.
102
 
 
103
 
        :type expect_body_tail: str
104
 
        :param expect_body_tail: a reply won't be sent until this string is
105
 
            received.
106
 
        """
107
 
        self._expect_body_tail = expect_body_tail
108
 
        self.host = None
109
 
        self.port = None
110
 
        self.received_bytes = ''
111
 
 
112
 
    def setUp(self):
113
 
        self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
114
 
        self._sock.bind(('127.0.0.1', 0))
115
 
        self.host, self.port = self._sock.getsockname()
116
 
        self._ready = threading.Event()
117
 
        self._thread = threading.Thread(target=self._accept_read_and_reply)
118
 
        self._thread.setDaemon(True)
119
 
        self._thread.start()
120
 
        self._ready.wait(5)
121
 
 
122
 
    def _accept_read_and_reply(self):
123
 
        self._sock.listen(1)
124
 
        self._ready.set()
125
 
        self._sock.settimeout(5)
126
 
        try:
127
 
            conn, address = self._sock.accept()
128
 
            # On win32, the accepted connection will be non-blocking to start
129
 
            # with because we're using settimeout.
130
 
            conn.setblocking(True)
131
 
            while not self.received_bytes.endswith(self._expect_body_tail):
132
 
                self.received_bytes += conn.recv(4096)
133
 
            conn.sendall('HTTP/1.1 200 OK\r\n')
134
 
        except socket.timeout:
135
 
            # Make sure the client isn't stuck waiting for us to e.g. accept.
136
 
            self._sock.close()
137
 
        except socket.error:
138
 
            # The client may have already closed the socket.
139
 
            pass
140
 
 
141
 
    def tearDown(self):
142
 
        try:
143
 
            self._sock.close()
144
 
        except socket.error:
145
 
            # We might have already closed it.  We don't care.
146
 
            pass
147
 
        self.host = None
148
 
        self.port = None
149
 
 
150
 
 
151
95
class TestHTTPServer(tests.TestCase):
152
96
    """Test the HTTP servers implementations."""
153
97
 
265
209
        self.assertRaises(errors.DependencyNotPresent, self._transport,
266
210
                          'https://launchpad.net')
267
211
 
268
 
class TestHttpConnections(object):
269
 
    """Test the http connections.
270
 
 
271
 
    This MUST be used by daughter classes that also inherit from
272
 
    TestCaseWithWebserver.
273
 
 
274
 
    We can't inherit directly from TestCaseWithWebserver or the
275
 
    test framework will try to create an instance which cannot
276
 
    run, its implementation being incomplete.
277
 
    """
278
 
 
279
 
    def setUp(self):
280
 
        TestCaseWithWebserver.setUp(self)
281
 
        self.build_tree(['xxx', 'foo/', 'foo/bar'], line_endings='binary',
282
 
                        transport=self.get_transport())
283
 
 
284
 
    def test_http_has(self):
285
 
        server = self.get_readonly_server()
286
 
        t = self._transport(server.get_url())
287
 
        self.assertEqual(t.has('foo/bar'), True)
288
 
        self.assertEqual(len(server.logs), 1)
289
 
        self.assertContainsRe(server.logs[0],
290
 
            r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "bzr/')
291
 
 
292
 
    def test_http_has_not_found(self):
293
 
        server = self.get_readonly_server()
294
 
        t = self._transport(server.get_url())
295
 
        self.assertEqual(t.has('not-found'), False)
296
 
        self.assertContainsRe(server.logs[1],
297
 
            r'"HEAD /not-found HTTP/1.." 404 - "-" "bzr/')
298
 
 
299
 
    def test_http_get(self):
300
 
        server = self.get_readonly_server()
301
 
        t = self._transport(server.get_url())
302
 
        fp = t.get('foo/bar')
303
 
        self.assertEqualDiff(
304
 
            fp.read(),
305
 
            'contents of foo/bar\n')
306
 
        self.assertEqual(len(server.logs), 1)
307
 
        self.assertTrue(server.logs[0].find(
308
 
            '"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s'
309
 
            % bzrlib.__version__) > -1)
310
 
 
311
 
    def test_get_smart_medium(self):
312
 
        # For HTTP, get_smart_medium should return the transport object.
313
 
        server = self.get_readonly_server()
314
 
        http_transport = self._transport(server.get_url())
315
 
        medium = http_transport.get_smart_medium()
316
 
        self.assertIs(medium, http_transport)
317
 
 
318
 
    def test_has_on_bogus_host(self):
319
 
        # Get a free address and don't 'accept' on it, so that we
320
 
        # can be sure there is no http handler there, but set a
321
 
        # reasonable timeout to not slow down tests too much.
322
 
        default_timeout = socket.getdefaulttimeout()
323
 
        try:
324
 
            socket.setdefaulttimeout(2)
325
 
            s = socket.socket()
326
 
            s.bind(('localhost', 0))
327
 
            t = self._transport('http://%s:%s/' % s.getsockname())
328
 
            self.assertRaises(errors.ConnectionError, t.has, 'foo/bar')
329
 
        finally:
330
 
            socket.setdefaulttimeout(default_timeout)
331
 
 
332
 
 
333
 
class TestHttpConnections_urllib(TestHttpConnections, TestCaseWithWebserver):
334
 
    """Test http connections with urllib"""
335
 
 
336
 
    _transport = HttpTransport_urllib
337
 
 
338
 
 
339
 
 
340
 
class TestHttpConnections_pycurl(TestWithTransport_pycurl,
341
 
                                 TestHttpConnections,
342
 
                                 TestCaseWithWebserver):
343
 
    """Test http connections with pycurl"""
344
 
 
345
 
 
346
212
class TestHttpTransportRegistration(tests.TestCase):
347
213
    """Test registrations of various http implementations"""
348
214
 
353
219
        self.assertIsInstance(t, HttpTransport_urllib)
354
220
 
355
221
 
356
 
class TestPost(object):
357
 
 
358
 
    def _test_post_body_is_received(self, scheme):
359
 
        server = RecordingServer(expect_body_tail='end-of-body')
360
 
        server.setUp()
361
 
        self.addCleanup(server.tearDown)
362
 
        url = '%s://%s:%s/' % (scheme, server.host, server.port)
363
 
        try:
364
 
            http_transport = get_transport(url)
365
 
        except errors.UnsupportedProtocol:
366
 
            raise tests.TestSkipped('%s not available' % scheme)
367
 
        code, response = http_transport._post('abc def end-of-body')
368
 
        self.assertTrue(
369
 
            server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
370
 
        self.assertTrue('content-length: 19\r' in server.received_bytes.lower())
371
 
        # The transport should not be assuming that the server can accept
372
 
        # chunked encoding the first time it connects, because HTTP/1.1, so we
373
 
        # check for the literal string.
374
 
        self.assertTrue(
375
 
            server.received_bytes.endswith('\r\n\r\nabc def end-of-body'))
376
 
 
377
 
 
378
 
class TestPost_urllib(tests.TestCase, TestPost):
379
 
    """TestPost for urllib implementation"""
380
 
 
381
 
    _transport = HttpTransport_urllib
382
 
 
383
 
    def test_post_body_is_received_urllib(self):
384
 
        self._test_post_body_is_received('http+urllib')
385
 
 
386
 
 
387
 
class TestPost_pycurl(TestWithTransport_pycurl, tests.TestCase, TestPost):
388
 
    """TestPost for pycurl implementation"""
389
 
 
390
 
    def test_post_body_is_received_pycurl(self):
391
 
        self._test_post_body_is_received('http+pycurl')
392
 
 
393
 
 
394
222
class TestRangeHeader(tests.TestCase):
395
223
    """Test range_header method"""
396
224
 
569
397
class TestRecordingServer(tests.TestCase):
570
398
 
571
399
    def test_create(self):
572
 
        server = RecordingServer(expect_body_tail=None)
 
400
        server = http_utils.RecordingServer(expect_body_tail=None)
573
401
        self.assertEqual('', server.received_bytes)
574
402
        self.assertEqual(None, server.host)
575
403
        self.assertEqual(None, server.port)
576
404
 
577
405
    def test_setUp_and_tearDown(self):
578
 
        server = RecordingServer(expect_body_tail=None)
 
406
        server = http_utils.RecordingServer(expect_body_tail=None)
579
407
        server.setUp()
580
408
        try:
581
409
            self.assertNotEqual(None, server.host)
586
414
        self.assertEqual(None, server.port)
587
415
 
588
416
    def test_send_receive_bytes(self):
589
 
        server = RecordingServer(expect_body_tail='c')
 
417
        server = http_utils.RecordingServer(expect_body_tail='c')
590
418
        server.setUp()
591
419
        self.addCleanup(server.tearDown)
592
420
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)