Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
77 changes: 29 additions & 48 deletions Lib/test/test_timeout.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,9 +5,6 @@
from test import support
from test.support import socket_helper

# This requires the 'network' resource as given on the regrtest command line.
skip_expected = not support.is_resource_enabled('network')

import time
import errno
import socket
Expand All @@ -29,10 +26,8 @@ class CreationTestCase(unittest.TestCase):
"""Test case for socket.gettimeout() and socket.settimeout()"""

def setUp(self):
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

def tearDown(self):
self.sock.close()
self.sock = self.enterContext(
socket.socket(socket.AF_INET, socket.SOCK_STREAM))

def testObjectCreation(self):
# Test Socket creation
Expand All @@ -53,10 +48,10 @@ def testFloatReturnValue(self):
def testReturnType(self):
# Test return type of gettimeout()
self.sock.settimeout(1)
self.assertEqual(type(self.sock.gettimeout()), type(1.0))
self.assertIs(type(self.sock.gettimeout()), float)

self.sock.settimeout(3.9)
self.assertEqual(type(self.sock.gettimeout()), type(1.0))
self.assertIs(type(self.sock.gettimeout()), float)

def testTypeCheck(self):
# Test type checking by settimeout()
Expand Down Expand Up @@ -116,8 +111,6 @@ class TimeoutTestCase(unittest.TestCase):
def setUp(self):
raise NotImplementedError()

tearDown = setUp

def _sock_operation(self, count, timeout, method, *args):
"""
Test the specified socket method.
Expand Down Expand Up @@ -145,19 +138,16 @@ class TCPTimeoutTestCase(TimeoutTestCase):
"""TCP test case for socket.socket() timeout functions"""

def setUp(self):
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock = self.enterContext(
socket.socket(socket.AF_INET, socket.SOCK_STREAM))
self.addr_remote = resolve_address('www.python.org.', 80)

def tearDown(self):
self.sock.close()

@unittest.skipIf(True, 'need to replace these hosts; see bpo-35518')
def testConnectTimeout(self):
# Testing connect timeout is tricky: we need to have IP connectivity
# to a host that silently drops our packets. We can't simulate this
# from Python because it's a function of the underlying TCP/IP stack.
# So, the following Snakebite host has been defined:
blackhole = resolve_address('blackhole.snakebite.net', 56666)
# So, the following port on the pythontest.net host has been defined:
blackhole = resolve_address('pythontest.net', 56666)

# Blackhole has been configured to silently drop any incoming packets.
# No RSTs (for TCP) or ICMP UNREACH (for UDP/ICMP) will be sent back
Expand All @@ -169,7 +159,7 @@ def testConnectTimeout(self):
# to firewalling or general network configuration. In order to improve
# our confidence in testing the blackhole, a corresponding 'whitehole'
# has also been set up using one port higher:
whitehole = resolve_address('whitehole.snakebite.net', 56667)
whitehole = resolve_address('pythontest.net', 56667)

# This address has been configured to immediately drop any incoming
# packets as well, but it does it respectfully with regards to the
Expand All @@ -183,35 +173,27 @@ def testConnectTimeout(self):
# timeframe).

# For the records, the whitehole/blackhole configuration has been set
# up using the 'pf' firewall (available on BSDs), using the following:
# up using the 'iptables' firewall, using the following rules:
#
# ext_if="bge0"
#
# blackhole_ip="35.8.247.6"
# whitehole_ip="35.8.247.6"
# blackhole_port="56666"
# whitehole_port="56667"
#
# block return in log quick on $ext_if proto { tcp udp } \
# from any to $whitehole_ip port $whitehole_port
# block drop in log quick on $ext_if proto { tcp udp } \
# from any to $blackhole_ip port $blackhole_port
# -A INPUT -p tcp --destination-port 56666 -j DROP
# -A INPUT -p udp --destination-port 56666 -j DROP
# -A INPUT -p tcp --destination-port 56667 -j REJECT
# -A INPUT -p udp --destination-port 56667 -j REJECT
#
# See https://github.com/python/psf-salt/blob/main/pillar/base/firewall/snakebite.sls
# for the current configuration.

skip = True
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
timeout = support.LOOPBACK_TIMEOUT
sock.settimeout(timeout)
try:
sock.connect((whitehole))
except TimeoutError:
pass
except OSError as err:
if err.errno == errno.ECONNREFUSED:
skip = False
finally:
sock.close()
del sock
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
try:
timeout = support.LOOPBACK_TIMEOUT
sock.settimeout(timeout)
sock.connect((whitehole))
except TimeoutError:
pass
except OSError as err:
if err.errno == errno.ECONNREFUSED:
skip = False

if skip:
self.skipTest(
Expand Down Expand Up @@ -278,10 +260,8 @@ class UDPTimeoutTestCase(TimeoutTestCase):
"""UDP test case for socket.socket() timeout functions"""

def setUp(self):
self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

def tearDown(self):
self.sock.close()
self.sock = self.enterContext(
socket.socket(socket.AF_INET, socket.SOCK_DGRAM))

def testRecvfromTimeout(self):
# Test recvfrom() timeout
Expand All @@ -292,6 +272,7 @@ def testRecvfromTimeout(self):

def setUpModule():
support.requires('network')
support.requires_working_socket(module=True)


if __name__ == "__main__":
Expand Down
Loading