unit tests for socket timeout feature import functools import unittest from test import support from test support import sockethelper import time import errno import socket functools lrucache def resolveaddresshost port with sockethelper transientinternethost return socket getaddrinfohost port socket afinet socket sockstream04 class creationtestcaseunittest testcase test socket creation test return value of gettimeout test return type of gettimeout test type checking by settimeout test range checking by settimeout test settimeout followed by setblocking test setblocking followed by settimeout there are a number of tests here trying to make sure that an operation doesn t take too much longer than expected but competing machine activity makes it inevitable that such tests will fail at times when fuzz was at 1 0 i tim routinely saw bogus failures on win2k and win98se boosting it to 2 0 helped a lot but isn t a real solution test the specified socket method the method is run at most count times and must raise a timeouterror within timeout self fuzz seconds these checks should account for timing unprecision tcp test case for socket socket timeout functions def setupself self sock socket socketsocket afinet socket sockstream self addrremote resolveaddress www python org 80 def teardownself self sock close def testconnecttimeoutself 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 tcpip stack so the following port on the pythontest net host has been defined blackhole resolveaddress pythontest net 56666 blackhole has been configured to silently drop any incoming packets no rsts for tcp or icmp unreach for udpicmp will be sent back to hosts that attempt to connect to this address which is exactly what we need to confidently test connect timeout however we want to prevent false positives it s not unreasonable to expect certain hosts may not be able to reach the blackhole due 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 resolveaddress 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 incoming protocol rsts are sent for tcp packets and icmp unreach is sent for udpicmp packets this means our attempts to connect to it should be met immediately with econnrefused the test case has been structured around this premise if we get an econnrefused from the whitehole we proceed with testing connect timeout against the blackhole if we don t we skip the test with a message about not getting the required rst from the whitehole within the required timeframe for the records the whiteholeblackhole configuration has been set up using the iptables firewall using the following rules a input p tcp destinationport 56666 j drop a input p udp destinationport 56666 j drop a input p tcp destinationport 56667 j reject a input p udp destinationport 56667 j reject see https github compythonpsfsaltblobmainpillarbasefirewallsnakebite sls for the current configuration skip true sock socket socketsocket afinet socket sockstream timeout support loopbacktimeout sock settimeouttimeout try sock connectwhitehole except timeouterror pass except oserror as err if err errno errno econnrefused skip false finally sock close del sock if skip self skiptest we didn t receive a connection reset rst packet from within seconds so we re unable to test connect timeout against the corresponding which is configured to silently drop packets format whitehole0 whitehole1 timeout blackhole0 blackhole1 all that hard work just to test if connect times out in 0 001s self addrremote blackhole with sockethelper transientinternetself addrremote0 self sockoperation1 0 001 connect self addrremote def testrecvtimeoutself test recv timeout with sockethelper transientinternetself addrremote0 self sock connectself addrremote self sockoperation1 1 5 recv 1024 def testaccepttimeoutself test accept timeout sockethelper bindportself sock self localhost self sock listen self sockoperation1 1 5 accept def testsendself test send timeout with socket socketsocket afinet socket sockstream as serv sockethelper bindportserv self localhost serv listen self sock connectserv getsockname send a lot of data in order to bypass buffering in the tcp stack self sockoperation100 1 5 send bx 200000 def testsendtoself test sendto timeout with socket socketsocket afinet socket sockstream as serv sockethelper bindportserv self localhost serv listen self sock connectserv getsockname the address argument is ignored since we already connected self sockoperation100 1 5 sendto bx 200000 serv getsockname def testsendallself test sendall timeout with socket socketsocket afinet socket sockstream as serv sockethelper bindportserv self localhost serv listen self sock connectserv getsockname send a lot of data in order to bypass buffering in the tcp stack self sockoperation100 1 5 sendall bx 200000 class udptimeouttestcasetimeouttestcase test recvfrom timeout prevent address already in use socket exceptions resolve an host port to an address we must perform name resolution before timeout tests otherwise it will be performed by connect test case for socket gettimeout and socket settimeout test socket creation test return value of gettimeout test return type of gettimeout test type checking by settimeout test range checking by settimeout test settimeout followed by setblocking test setblocking followed by settimeout there are a number of tests here trying to make sure that an operation doesn t take too much longer than expected but competing machine activity makes it inevitable that such tests will fail at times when fuzz was at 1 0 i tim routinely saw bogus failures on win2k and win98se boosting it to 2 0 helped a lot but isn t a real solution test the specified socket method the method is run at most count times and must raise a timeouterror within timeout self fuzz seconds these checks should account for timing unprecision tcp test case for socket socket timeout functions 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 port on the pythontest net host has been defined blackhole has been configured to silently drop any incoming packets no rsts for tcp or icmp unreach for udp icmp will be sent back to hosts that attempt to connect to this address which is exactly what we need to confidently test connect timeout however we want to prevent false positives it s not unreasonable to expect certain hosts may not be able to reach the blackhole due 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 this address has been configured to immediately drop any incoming packets as well but it does it respectfully with regards to the incoming protocol rsts are sent for tcp packets and icmp unreach is sent for udp icmp packets this means our attempts to connect to it should be met immediately with econnrefused the test case has been structured around this premise if we get an econnrefused from the whitehole we proceed with testing connect timeout against the blackhole if we don t we skip the test with a message about not getting the required rst from the whitehole within the required timeframe for the records the whitehole blackhole configuration has been set up using the iptables firewall using the following rules 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 all that hard work just to test if connect times out in 0 001s test recv timeout test accept timeout test send timeout send a lot of data in order to bypass buffering in the tcp stack test sendto timeout the address argument is ignored since we already connected test sendall timeout send a lot of data in order to bypass buffering in the tcp stack udp test case for socket socket timeout functions test recvfrom timeout prevent address already in use socket exceptions
import functools
import unittest
from test import support
from test.support import socket_helper

import time
import errno
import socket


@functools.lru_cache()
def resolve_address(host, port):
    
    with socket_helper.transient_internet(host):
        return socket.getaddrinfo(host, port, socket.AF_INET,
                                  socket.SOCK_STREAM)[0][4]


class CreationTestCase(unittest.TestCase):
    

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

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

    def testObjectCreation(self):
        
        self.assertEqual(self.sock.gettimeout(), None,
                         "timeout not disabled by default")

    def testFloatReturnValue(self):
        
        self.sock.settimeout(7.345)
        self.assertEqual(self.sock.gettimeout(), 7.345)

        self.sock.settimeout(3)
        self.assertEqual(self.sock.gettimeout(), 3)

        self.sock.settimeout(None)
        self.assertEqual(self.sock.gettimeout(), None)

    def testReturnType(self):
        
        self.sock.settimeout(1)
        self.assertIs(type(self.sock.gettimeout()), float)

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

    def testTypeCheck(self):
        
        self.sock.settimeout(0)
        self.sock.settimeout(0)
        self.sock.settimeout(0.0)
        self.sock.settimeout(None)
        self.assertRaises(TypeError, self.sock.settimeout, "")
        self.assertRaises(TypeError, self.sock.settimeout, "")
        self.assertRaises(TypeError, self.sock.settimeout, ())
        self.assertRaises(TypeError, self.sock.settimeout, [])
        self.assertRaises(TypeError, self.sock.settimeout, {})
        self.assertRaises(TypeError, self.sock.settimeout, 0j)

    def testRangeCheck(self):
        
        self.assertRaises(ValueError, self.sock.settimeout, -1)
        self.assertRaises(ValueError, self.sock.settimeout, -1)
        self.assertRaises(ValueError, self.sock.settimeout, -1.0)

    def testTimeoutThenBlocking(self):
        
        self.sock.settimeout(10)
        self.sock.setblocking(True)
        self.assertEqual(self.sock.gettimeout(), None)
        self.sock.setblocking(False)
        self.assertEqual(self.sock.gettimeout(), 0.0)

        self.sock.settimeout(10)
        self.sock.setblocking(False)
        self.assertEqual(self.sock.gettimeout(), 0.0)
        self.sock.setblocking(True)
        self.assertEqual(self.sock.gettimeout(), None)

    def testBlockingThenTimeout(self):
        
        self.sock.setblocking(False)
        self.sock.settimeout(1)
        self.assertEqual(self.sock.gettimeout(), 1)

        self.sock.setblocking(True)
        self.sock.settimeout(1)
        self.assertEqual(self.sock.gettimeout(), 1)


class TimeoutTestCase(unittest.TestCase):
    
    
    
    
    
    
    fuzz = 2.0

    localhost = socket_helper.HOST

    def setUp(self):
        raise NotImplementedError()

    tearDown = setUp

    def _sock_operation(self, count, timeout, method, *args):
        
        self.sock.settimeout(timeout)
        method = getattr(self.sock, method)
        for i in range(count):
            t1 = time.monotonic()
            try:
                method(*args)
            except TimeoutError as e:
                delta = time.monotonic() - t1
                break
        else:
            self.fail('TimeoutError was not raised')
        
        self.assertLess(delta, timeout + self.fuzz)
        self.assertGreater(delta, timeout - 1.0)


class TCPTimeoutTestCase(TimeoutTestCase):
    

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

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

    def testConnectTimeout(self):
        
        
        
        
        blackhole = resolve_address('pythontest.net', 56666)

        
        
        
        

        
        
        
        
        
        whitehole = resolve_address('pythontest.net', 56667)

        
        
        
        
        
        
        
        
        
        

        
        
        
        
        
        
        
        
        
        

        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

        if skip:
            self.skipTest(
                "We didn't receive a connection reset (RST) packet from "
                "{}:{} within {} seconds, so we're unable to test connect "
                "timeout against the corresponding {}:{} (which is "
                "configured to silently drop packets)."
                    .format(
                        whitehole[0],
                        whitehole[1],
                        timeout,
                        blackhole[0],
                        blackhole[1],
                    )
            )

        
        self.addr_remote = blackhole
        with socket_helper.transient_internet(self.addr_remote[0]):
            self._sock_operation(1, 0.001, 'connect', self.addr_remote)

    def testRecvTimeout(self):
        
        with socket_helper.transient_internet(self.addr_remote[0]):
            self.sock.connect(self.addr_remote)
            self._sock_operation(1, 1.5, 'recv', 1024)

    def testAcceptTimeout(self):
        
        socket_helper.bind_port(self.sock, self.localhost)
        self.sock.listen()
        self._sock_operation(1, 1.5, 'accept')

    def testSend(self):
        
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as serv:
            socket_helper.bind_port(serv, self.localhost)
            serv.listen()
            self.sock.connect(serv.getsockname())
            
            self._sock_operation(100, 1.5, 'send', b"X" * 200000)

    def testSendto(self):
        
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as serv:
            socket_helper.bind_port(serv, self.localhost)
            serv.listen()
            self.sock.connect(serv.getsockname())
            
            self._sock_operation(100, 1.5, 'sendto', b"X" * 200000,
                                 serv.getsockname())

    def testSendall(self):
        
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as serv:
            socket_helper.bind_port(serv, self.localhost)
            serv.listen()
            self.sock.connect(serv.getsockname())
            
            self._sock_operation(100, 1.5, 'sendall', b"X" * 200000)


class UDPTimeoutTestCase(TimeoutTestCase):
    

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

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

    def testRecvfromTimeout(self):
        
        
        socket_helper.bind_port(self.sock, self.localhost)
        self._sock_operation(1, 1.5, 'recvfrom', 1024)


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


if __name__ == "__main__":
    unittest.main()