Merged revisions 84597-84599 via svnmerge from

svn+ssh://pythondev@svn.python.org/python/branches/py3k

........
  r84597 | antoine.pitrou | 2010-09-07 22:42:19 +0200 (mar., 07 sept. 2010) | 5 lines

  Issue #8574: better implementation of test.support.transient_internet().
  Original patch by Victor.
........
  r84598 | antoine.pitrou | 2010-09-07 23:05:49 +0200 (mar., 07 sept. 2010) | 6 lines

  Issue #9792: In case of connection failure, socket.create_connection()
  would swallow the exception and raise a new one, making it impossible
  to fetch the original errno, or to filter timeout errors.  Now the
  original error is re-raised.
........
  r84599 | antoine.pitrou | 2010-09-07 23:09:09 +0200 (mar., 07 sept. 2010) | 4 lines

  Improve transient_internet() again to detect more network errors,
  and use it in test_robotparser. Fixes #8574.
........
This commit is contained in:
Antoine Pitrou 2010-09-07 21:22:56 +00:00
parent 824cf253e5
commit 4d7979be72
6 changed files with 134 additions and 46 deletions

View file

@ -14,6 +14,7 @@ import queue
import sys
import os
import array
import contextlib
from weakref import proxy
import signal
@ -1026,12 +1027,48 @@ class BasicTCPTest2(NetworkConnectionTest, BasicTCPTest):
class NetworkConnectionNoServer(unittest.TestCase):
def testWithoutServer(self):
class MockSocket(socket.socket):
def connect(self, *args):
raise socket.timeout('timed out')
@contextlib.contextmanager
def mocked_socket_module(self):
"""Return a socket which times out on connect"""
old_socket = socket.socket
socket.socket = self.MockSocket
try:
yield
finally:
socket.socket = old_socket
def test_connect(self):
port = support.find_unused_port()
self.assertRaises(
socket.error,
lambda: socket.create_connection((HOST, port))
)
cli = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
cli.connect((HOST, port))
except socket.error as err:
self.assertEqual(err.errno, errno.ECONNREFUSED)
else:
self.fail("socket.error not raised")
def test_create_connection(self):
# Issue #9792: errors raised by create_connection() should have
# a proper errno attribute.
port = support.find_unused_port()
try:
socket.create_connection((HOST, port))
except socket.error as err:
self.assertEqual(err.errno, errno.ECONNREFUSED)
else:
self.fail("socket.error not raised")
def test_create_connection_timeout(self):
# Issue #9792: create_connection() should not recast timeout errors
# as generic socket errors.
with self.mocked_socket_module():
with self.assertRaises(socket.timeout):
socket.create_connection((HOST, 1234))
class NetworkConnectionAttributesTest(SocketTCPTest, ThreadableTest):