test_timeout.py 5.76 KB
Newer Older
1
"""Unit tests for socket timeout feature."""
2 3

import unittest
4
import test_support
5 6 7 8

import time
import socket

9

10
class CreationTestCase(unittest.TestCase):
11
    """Test case for socket.gettimeout() and socket.settimeout()"""
12

13
    def setUp(self):
14
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
15 16

    def tearDown(self):
17
        self.sock.close()
18 19 20

    def testObjectCreation(self):
        "Test Socket creation"
21 22
        self.assertEqual(self.sock.gettimeout(), None,
                         "timeout not disabled by default")
23 24

    def testFloatReturnValue(self):
25 26 27
        "Test return value of gettimeout()"
        self.sock.settimeout(7.345)
        self.assertEqual(self.sock.gettimeout(), 7.345)
28

29 30
        self.sock.settimeout(3)
        self.assertEqual(self.sock.gettimeout(), 3)
31

32 33
        self.sock.settimeout(None)
        self.assertEqual(self.sock.gettimeout(), None)
34

35 36 37 38
    def testReturnType(self):
        "Test return type of gettimeout()"
        self.sock.settimeout(1)
        self.assertEqual(type(self.sock.gettimeout()), type(1.0))
39

40 41
        self.sock.settimeout(3.9)
        self.assertEqual(type(self.sock.gettimeout()), type(1.0))
42 43

    def testTypeCheck(self):
44 45 46 47 48 49 50 51 52 53 54
        "Test type checking by settimeout()"
        self.sock.settimeout(0)
        self.sock.settimeout(0L)
        self.sock.settimeout(0.0)
        self.sock.settimeout(None)
        self.assertRaises(TypeError, self.sock.settimeout, "")
        self.assertRaises(TypeError, self.sock.settimeout, u"")
        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)
55 56

    def testRangeCheck(self):
57 58 59 60
        "Test range checking by settimeout()"
        self.assertRaises(ValueError, self.sock.settimeout, -1)
        self.assertRaises(ValueError, self.sock.settimeout, -1L)
        self.assertRaises(ValueError, self.sock.settimeout, -1.0)
61

62
    def testTimeoutThenBlocking(self):
63 64 65 66 67
        "Test settimeout() followed by setblocking()"
        self.sock.settimeout(10)
        self.sock.setblocking(1)
        self.assertEqual(self.sock.gettimeout(), None)
        self.sock.setblocking(0)
68
        self.assertEqual(self.sock.gettimeout(), 0.0)
69 70 71

        self.sock.settimeout(10)
        self.sock.setblocking(0)
72
        self.assertEqual(self.sock.gettimeout(), 0.0)
73 74
        self.sock.setblocking(1)
        self.assertEqual(self.sock.gettimeout(), None)
75 76

    def testBlockingThenTimeout(self):
77 78 79 80
        "Test setblocking() followed by settimeout()"
        self.sock.setblocking(0)
        self.sock.settimeout(1)
        self.assertEqual(self.sock.gettimeout(), 1)
81

82 83 84
        self.sock.setblocking(1)
        self.sock.settimeout(1)
        self.assertEqual(self.sock.gettimeout(), 1)
85

86

87
class TimeoutTestCase(unittest.TestCase):
88 89 90
    """Test case for socket.socket() timeout functions"""

    fuzz = 1.0
91

92
    def setUp(self):
93 94 95
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.addr_remote = ('www.google.com', 80)
        self.addr_local  = ('127.0.0.1', 25339)
96 97

    def tearDown(self):
98
        self.sock.close()
99 100 101 102

    def testConnectTimeout(self):
        "Test connect() timeout"
        _timeout = 0.02
103
        self.sock.settimeout(_timeout)
104 105

        _t1 = time.time()
106 107
        self.failUnlessRaises(socket.error, self.sock.connect,
                self.addr_remote)
108 109 110
        _t2 = time.time()

        _delta = abs(_t1 - _t2)
111 112 113
        self.assert_(_delta < _timeout + self.fuzz,
                     "timeout (%g) is %g seconds more than expected (%g)"
                     %(_delta, self.fuzz, _timeout))
114 115 116 117

    def testRecvTimeout(self):
        "Test recv() timeout"
        _timeout = 0.02
118 119
        self.sock.connect(self.addr_remote)
        self.sock.settimeout(_timeout)
120 121

        _t1 = time.time()
122
        self.failUnlessRaises(socket.error, self.sock.recv, 1024)
123 124 125
        _t2 = time.time()

        _delta = abs(_t1 - _t2)
126 127 128
        self.assert_(_delta < _timeout + self.fuzz,
                     "timeout (%g) is %g seconds more than expected (%g)"
                     %(_delta, self.fuzz, _timeout))
129 130

    def testAcceptTimeout(self):
131
        "Test accept() timeout"
132
        _timeout = 2
133 134 135
        self.sock.settimeout(_timeout)
        self.sock.bind(self.addr_local)
        self.sock.listen(5)
136 137

        _t1 = time.time()
138
        self.failUnlessRaises(socket.error, self.sock.accept)
139 140 141
        _t2 = time.time()

        _delta = abs(_t1 - _t2)
142 143 144
        self.assert_(_delta < _timeout + self.fuzz,
                     "timeout (%g) is %g seconds more than expected (%g)"
                     %(_delta, self.fuzz, _timeout))
145 146

    def testRecvfromTimeout(self):
147
        "Test recvfrom() timeout"
148
        _timeout = 2
149 150 151
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.sock.settimeout(_timeout)
        self.sock.bind(self.addr_local)
152 153

        _t1 = time.time()
154
        self.failUnlessRaises(socket.error, self.sock.recvfrom, 8192)
155 156 157
        _t2 = time.time()

        _delta = abs(_t1 - _t2)
158 159 160
        self.assert_(_delta < _timeout + self.fuzz,
                     "timeout (%g) is %g seconds more than expected (%g)"
                     %(_delta, self.fuzz, _timeout))
161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177

    def testSend(self):
        "Test send() timeout"
        # couldn't figure out how to test it
        pass

    def testSendto(self):
        "Test sendto() timeout"
        # couldn't figure out how to test it
        pass

    def testSendall(self):
        "Test sendall() timeout"
        # couldn't figure out how to test it
        pass


178
def main():
179
    suite = unittest.TestSuite()
180 181 182
    suite.addTest(unittest.makeSuite(CreationTestCase))
    suite.addTest(unittest.makeSuite(TimeoutTestCase))
    test_support.run_suite(suite)
183 184

if __name__ == "__main__":
185
    main()