Rev 3118: Further refactoring. in file:///v/home/vila/src/bzr/bugs/175524/
Vincent Ladeuil
v.ladeuil+lp at free.fr
Wed Dec 19 09:29:27 GMT 2007
At file:///v/home/vila/src/bzr/bugs/175524/
------------------------------------------------------------
revno: 3118
revision-id:v.ladeuil+lp at free.fr-20071219092921-3oxvnp5ptfhz8v42
parent: v.ladeuil+lp at free.fr-20071219074629-nu5z8y56lp9o3m7f
committer: Vincent Ladeuil <v.ladeuil+lp at free.fr>
branch nick: 175524
timestamp: Wed 2007-12-19 10:29:21 +0100
message:
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).
modified:
bzrlib/tests/http_utils.py HTTPTestUtil.py-20050914180604-247d3aafb7a43343
bzrlib/tests/test_http.py testhttp.py-20051018020158-b2eef6e867c514d9
bzrlib/tests/test_http_implementations.py test_http_implementa-20071218210003-65nh81gglcfvurw6-1
-------------- next part --------------
=== modified file 'bzrlib/tests/http_utils.py'
--- a/bzrlib/tests/http_utils.py 2007-12-11 14:26:18 +0000
+++ b/bzrlib/tests/http_utils.py 2007-12-19 09:29:21 +0000
@@ -21,6 +21,7 @@
import re
import sha
import socket
+import threading
import time
import urllib2
import urlparse
@@ -561,3 +562,60 @@
self.init_proxy_auth()
+class RecordingServer(object):
+ """A fake HTTP server.
+
+ It records the bytes sent to it, and replies with a 200.
+ """
+
+ def __init__(self, expect_body_tail=None):
+ """Constructor.
+
+ :type expect_body_tail: str
+ :param expect_body_tail: a reply won't be sent until this string is
+ received.
+ """
+ self._expect_body_tail = expect_body_tail
+ self.host = None
+ self.port = None
+ self.received_bytes = ''
+
+ def setUp(self):
+ self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ self._sock.bind(('127.0.0.1', 0))
+ self.host, self.port = self._sock.getsockname()
+ self._ready = threading.Event()
+ self._thread = threading.Thread(target=self._accept_read_and_reply)
+ self._thread.setDaemon(True)
+ self._thread.start()
+ self._ready.wait(5)
+
+ def _accept_read_and_reply(self):
+ self._sock.listen(1)
+ self._ready.set()
+ self._sock.settimeout(5)
+ try:
+ conn, address = self._sock.accept()
+ # On win32, the accepted connection will be non-blocking to start
+ # with because we're using settimeout.
+ conn.setblocking(True)
+ while not self.received_bytes.endswith(self._expect_body_tail):
+ self.received_bytes += conn.recv(4096)
+ conn.sendall('HTTP/1.1 200 OK\r\n')
+ except socket.timeout:
+ # Make sure the client isn't stuck waiting for us to e.g. accept.
+ self._sock.close()
+ except socket.error:
+ # The client may have already closed the socket.
+ pass
+
+ def tearDown(self):
+ try:
+ self._sock.close()
+ except socket.error:
+ # We might have already closed it. We don't care.
+ pass
+ self.host = None
+ self.port = None
+
+
=== modified file 'bzrlib/tests/test_http.py'
--- a/bzrlib/tests/test_http.py 2007-12-19 07:46:29 +0000
+++ b/bzrlib/tests/test_http.py 2007-12-19 09:29:21 +0000
@@ -39,6 +39,7 @@
)
from bzrlib.tests import (
http_server,
+ http_utils,
TestCase,
TestUIFactory,
TestSkipped,
@@ -91,63 +92,6 @@
self.credentials.append([realm, host, username, password])
-class RecordingServer(object):
- """A fake HTTP server.
-
- It records the bytes sent to it, and replies with a 200.
- """
-
- def __init__(self, expect_body_tail=None):
- """Constructor.
-
- :type expect_body_tail: str
- :param expect_body_tail: a reply won't be sent until this string is
- received.
- """
- self._expect_body_tail = expect_body_tail
- self.host = None
- self.port = None
- self.received_bytes = ''
-
- def setUp(self):
- self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- self._sock.bind(('127.0.0.1', 0))
- self.host, self.port = self._sock.getsockname()
- self._ready = threading.Event()
- self._thread = threading.Thread(target=self._accept_read_and_reply)
- self._thread.setDaemon(True)
- self._thread.start()
- self._ready.wait(5)
-
- def _accept_read_and_reply(self):
- self._sock.listen(1)
- self._ready.set()
- self._sock.settimeout(5)
- try:
- conn, address = self._sock.accept()
- # On win32, the accepted connection will be non-blocking to start
- # with because we're using settimeout.
- conn.setblocking(True)
- while not self.received_bytes.endswith(self._expect_body_tail):
- self.received_bytes += conn.recv(4096)
- conn.sendall('HTTP/1.1 200 OK\r\n')
- except socket.timeout:
- # Make sure the client isn't stuck waiting for us to e.g. accept.
- self._sock.close()
- except socket.error:
- # The client may have already closed the socket.
- pass
-
- def tearDown(self):
- try:
- self._sock.close()
- except socket.error:
- # We might have already closed it. We don't care.
- pass
- self.host = None
- self.port = None
-
-
class TestHTTPServer(tests.TestCase):
"""Test the HTTP servers implementations."""
@@ -265,84 +209,6 @@
self.assertRaises(errors.DependencyNotPresent, self._transport,
'https://launchpad.net')
-class TestHttpConnections(object):
- """Test the http connections.
-
- This MUST be used by daughter classes that also inherit from
- TestCaseWithWebserver.
-
- We can't inherit directly from TestCaseWithWebserver or the
- test framework will try to create an instance which cannot
- run, its implementation being incomplete.
- """
-
- def setUp(self):
- TestCaseWithWebserver.setUp(self)
- self.build_tree(['xxx', 'foo/', 'foo/bar'], line_endings='binary',
- transport=self.get_transport())
-
- def test_http_has(self):
- server = self.get_readonly_server()
- t = self._transport(server.get_url())
- self.assertEqual(t.has('foo/bar'), True)
- self.assertEqual(len(server.logs), 1)
- self.assertContainsRe(server.logs[0],
- r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "bzr/')
-
- def test_http_has_not_found(self):
- server = self.get_readonly_server()
- t = self._transport(server.get_url())
- self.assertEqual(t.has('not-found'), False)
- self.assertContainsRe(server.logs[1],
- r'"HEAD /not-found HTTP/1.." 404 - "-" "bzr/')
-
- def test_http_get(self):
- server = self.get_readonly_server()
- t = self._transport(server.get_url())
- fp = t.get('foo/bar')
- self.assertEqualDiff(
- fp.read(),
- 'contents of foo/bar\n')
- self.assertEqual(len(server.logs), 1)
- self.assertTrue(server.logs[0].find(
- '"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s'
- % bzrlib.__version__) > -1)
-
- def test_get_smart_medium(self):
- # For HTTP, get_smart_medium should return the transport object.
- server = self.get_readonly_server()
- http_transport = self._transport(server.get_url())
- medium = http_transport.get_smart_medium()
- self.assertIs(medium, http_transport)
-
- def test_has_on_bogus_host(self):
- # Get a free address and don't 'accept' on it, so that we
- # can be sure there is no http handler there, but set a
- # reasonable timeout to not slow down tests too much.
- default_timeout = socket.getdefaulttimeout()
- try:
- socket.setdefaulttimeout(2)
- s = socket.socket()
- s.bind(('localhost', 0))
- t = self._transport('http://%s:%s/' % s.getsockname())
- self.assertRaises(errors.ConnectionError, t.has, 'foo/bar')
- finally:
- socket.setdefaulttimeout(default_timeout)
-
-
-class TestHttpConnections_urllib(TestHttpConnections, TestCaseWithWebserver):
- """Test http connections with urllib"""
-
- _transport = HttpTransport_urllib
-
-
-
-class TestHttpConnections_pycurl(TestWithTransport_pycurl,
- TestHttpConnections,
- TestCaseWithWebserver):
- """Test http connections with pycurl"""
-
-
class TestHttpTransportRegistration(tests.TestCase):
"""Test registrations of various http implementations"""
@@ -353,44 +219,6 @@
self.assertIsInstance(t, HttpTransport_urllib)
-class TestPost(object):
-
- def _test_post_body_is_received(self, scheme):
- server = RecordingServer(expect_body_tail='end-of-body')
- server.setUp()
- self.addCleanup(server.tearDown)
- url = '%s://%s:%s/' % (scheme, server.host, server.port)
- try:
- http_transport = get_transport(url)
- except errors.UnsupportedProtocol:
- raise tests.TestSkipped('%s not available' % scheme)
- code, response = http_transport._post('abc def end-of-body')
- self.assertTrue(
- server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
- self.assertTrue('content-length: 19\r' in server.received_bytes.lower())
- # The transport should not be assuming that the server can accept
- # chunked encoding the first time it connects, because HTTP/1.1, so we
- # check for the literal string.
- self.assertTrue(
- server.received_bytes.endswith('\r\n\r\nabc def end-of-body'))
-
-
-class TestPost_urllib(tests.TestCase, TestPost):
- """TestPost for urllib implementation"""
-
- _transport = HttpTransport_urllib
-
- def test_post_body_is_received_urllib(self):
- self._test_post_body_is_received('http+urllib')
-
-
-class TestPost_pycurl(TestWithTransport_pycurl, tests.TestCase, TestPost):
- """TestPost for pycurl implementation"""
-
- def test_post_body_is_received_pycurl(self):
- self._test_post_body_is_received('http+pycurl')
-
-
class TestRangeHeader(tests.TestCase):
"""Test range_header method"""
@@ -569,13 +397,13 @@
class TestRecordingServer(tests.TestCase):
def test_create(self):
- server = RecordingServer(expect_body_tail=None)
+ server = http_utils.RecordingServer(expect_body_tail=None)
self.assertEqual('', server.received_bytes)
self.assertEqual(None, server.host)
self.assertEqual(None, server.port)
def test_setUp_and_tearDown(self):
- server = RecordingServer(expect_body_tail=None)
+ server = http_utils.RecordingServer(expect_body_tail=None)
server.setUp()
try:
self.assertNotEqual(None, server.host)
@@ -586,7 +414,7 @@
self.assertEqual(None, server.port)
def test_send_receive_bytes(self):
- server = RecordingServer(expect_body_tail='c')
+ server = http_utils.RecordingServer(expect_body_tail='c')
server.setUp()
self.addCleanup(server.tearDown)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
=== modified file 'bzrlib/tests/test_http_implementations.py'
--- a/bzrlib/tests/test_http_implementations.py 2007-12-19 07:46:29 +0000
+++ b/bzrlib/tests/test_http_implementations.py 2007-12-19 09:29:21 +0000
@@ -25,12 +25,17 @@
protocol implementation to guarantee the robustness of our transports.
"""
+import socket
+
+import bzrlib
from bzrlib import (
errors,
tests,
+ transport,
)
from bzrlib.tests import (
http_server,
+ http_utils,
)
from bzrlib.transport.http._urllib import HttpTransport_urllib
@@ -105,3 +110,83 @@
server.tearDown()
+class TestHttpConnections(http_utils.TestCaseWithWebserver):
+ """Test the http connections."""
+
+ def setUp(self):
+ http_utils.TestCaseWithWebserver.setUp(self)
+ self.build_tree(['foo/', 'foo/bar'], line_endings='binary',
+ transport=self.get_transport())
+
+ def test_http_has(self):
+ server = self.get_readonly_server()
+ t = self._transport(server.get_url())
+ self.assertEqual(t.has('foo/bar'), True)
+ self.assertEqual(len(server.logs), 1)
+ self.assertContainsRe(server.logs[0],
+ r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "bzr/')
+
+ def test_http_has_not_found(self):
+ server = self.get_readonly_server()
+ t = self._transport(server.get_url())
+ self.assertEqual(t.has('not-found'), False)
+ self.assertContainsRe(server.logs[1],
+ r'"HEAD /not-found HTTP/1.." 404 - "-" "bzr/')
+
+ def test_http_get(self):
+ server = self.get_readonly_server()
+ t = self._transport(server.get_url())
+ fp = t.get('foo/bar')
+ self.assertEqualDiff(
+ fp.read(),
+ 'contents of foo/bar\n')
+ self.assertEqual(len(server.logs), 1)
+ self.assertTrue(server.logs[0].find(
+ '"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s'
+ % bzrlib.__version__) > -1)
+
+ def test_get_smart_medium(self):
+ # For HTTP, get_smart_medium should return the transport object.
+ server = self.get_readonly_server()
+ http_transport = self._transport(server.get_url())
+ medium = http_transport.get_smart_medium()
+ self.assertIs(medium, http_transport)
+
+ def test_has_on_bogus_host(self):
+ # Get a free address and don't 'accept' on it, so that we
+ # can be sure there is no http handler there, but set a
+ # reasonable timeout to not slow down tests too much.
+ default_timeout = socket.getdefaulttimeout()
+ try:
+ socket.setdefaulttimeout(2)
+ s = socket.socket()
+ s.bind(('localhost', 0))
+ t = self._transport('http://%s:%s/' % s.getsockname())
+ self.assertRaises(errors.ConnectionError, t.has, 'foo/bar')
+ finally:
+ socket.setdefaulttimeout(default_timeout)
+
+
+class TestPost(tests.TestCase):
+
+ def test_post_body_is_received(self):
+ server = http_utils.RecordingServer(expect_body_tail='end-of-body')
+ server.setUp()
+ self.addCleanup(server.tearDown)
+ scheme = self._qualified_prefix
+ url = '%s://%s:%s/' % (scheme, server.host, server.port)
+ try:
+ http_transport = transport.get_transport(url)
+ except errors.UnsupportedProtocol:
+ raise tests.TestSkipped('%s not available' % scheme)
+ code, response = http_transport._post('abc def end-of-body')
+ self.assertTrue(
+ server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
+ self.assertTrue('content-length: 19\r' in server.received_bytes.lower())
+ # The transport should not be assuming that the server can accept
+ # chunked encoding the first time it connects, because HTTP/1.1, so we
+ # check for the literal string.
+ self.assertTrue(
+ server.received_bytes.endswith('\r\n\r\nabc def end-of-body'))
+
+
More information about the bazaar-commits
mailing list