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

import unittest
4
from test import test_support
5

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

9 10 11
import time
import socket

12

13
class CreationTestCase(unittest.TestCase):
14
    """Test case for socket.gettimeout() and socket.settimeout()"""
15

16
    def setUp(self):
17
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
18 19

    def tearDown(self):
20
        self.sock.close()
21 22

    def testObjectCreation(self):
23
        # Test Socket creation
24 25
        self.assertEqual(self.sock.gettimeout(), None,
                         "timeout not disabled by default")
26 27

    def testFloatReturnValue(self):
28
        # Test return value of gettimeout()
29 30
        self.sock.settimeout(7.345)
        self.assertEqual(self.sock.gettimeout(), 7.345)
31

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

35 36
        self.sock.settimeout(None)
        self.assertEqual(self.sock.gettimeout(), None)
37

38
    def testReturnType(self):
39
        # Test return type of gettimeout()
40 41
        self.sock.settimeout(1)
        self.assertEqual(type(self.sock.gettimeout()), type(1.0))
42

43 44
        self.sock.settimeout(3.9)
        self.assertEqual(type(self.sock.gettimeout()), type(1.0))
45 46

    def testTypeCheck(self):
47
        # Test type checking by settimeout()
48 49 50 51 52 53 54 55 56 57
        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)
58 59

    def testRangeCheck(self):
60
        # Test range checking by settimeout()
61 62 63
        self.assertRaises(ValueError, self.sock.settimeout, -1)
        self.assertRaises(ValueError, self.sock.settimeout, -1L)
        self.assertRaises(ValueError, self.sock.settimeout, -1.0)
64

65
    def testTimeoutThenBlocking(self):
66
        # Test settimeout() followed by setblocking()
67 68 69 70
        self.sock.settimeout(10)
        self.sock.setblocking(1)
        self.assertEqual(self.sock.gettimeout(), None)
        self.sock.setblocking(0)
71
        self.assertEqual(self.sock.gettimeout(), 0.0)
72 73 74

        self.sock.settimeout(10)
        self.sock.setblocking(0)
75
        self.assertEqual(self.sock.gettimeout(), 0.0)
76 77
        self.sock.setblocking(1)
        self.assertEqual(self.sock.gettimeout(), None)
78 79

    def testBlockingThenTimeout(self):
80
        # Test setblocking() followed by settimeout()
81 82 83
        self.sock.setblocking(0)
        self.sock.settimeout(1)
        self.assertEqual(self.sock.gettimeout(), 1)
84

85 86 87
        self.sock.setblocking(1)
        self.sock.settimeout(1)
        self.assertEqual(self.sock.gettimeout(), 1)
88

89

90
class TimeoutTestCase(unittest.TestCase):
91 92
    """Test case for socket.socket() timeout functions"""

93 94 95 96 97 98 99
    # 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.
    fuzz = 2.0
100

101
    def setUp(self):
102
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
103
        self.addr_remote = ('www.python.org', 80)
104
        self.addr_local  = ('127.0.0.1', 25339)
105 106

    def tearDown(self):
107
        self.sock.close()
108 109

    def testConnectTimeout(self):
110
        # Test connect() timeout
111
        _timeout = 0.001
112
        self.sock.settimeout(_timeout)
113 114

        _t1 = time.time()
115 116
        self.failUnlessRaises(socket.error, self.sock.connect,
                self.addr_remote)
117 118 119
        _t2 = time.time()

        _delta = abs(_t1 - _t2)
120
        self.assert_(_delta < _timeout + self.fuzz,
121
                     "timeout (%g) is more than %g seconds more than expected (%g)"
122
                     %(_delta, self.fuzz, _timeout))
123 124

    def testRecvTimeout(self):
125
        # Test recv() timeout
126
        _timeout = 0.02
127 128
        self.sock.connect(self.addr_remote)
        self.sock.settimeout(_timeout)
129 130

        _t1 = time.time()
131
        self.failUnlessRaises(socket.error, self.sock.recv, 1024)
132 133 134
        _t2 = time.time()

        _delta = abs(_t1 - _t2)
135 136 137
        self.assert_(_delta < _timeout + self.fuzz,
                     "timeout (%g) is %g seconds more than expected (%g)"
                     %(_delta, self.fuzz, _timeout))
138 139

    def testAcceptTimeout(self):
140
        # Test accept() timeout
141
        _timeout = 2
142 143 144
        self.sock.settimeout(_timeout)
        self.sock.bind(self.addr_local)
        self.sock.listen(5)
145 146

        _t1 = time.time()
147
        self.failUnlessRaises(socket.error, self.sock.accept)
148 149 150
        _t2 = time.time()

        _delta = abs(_t1 - _t2)
151 152 153
        self.assert_(_delta < _timeout + self.fuzz,
                     "timeout (%g) is %g seconds more than expected (%g)"
                     %(_delta, self.fuzz, _timeout))
154 155

    def testRecvfromTimeout(self):
156
        # Test recvfrom() timeout
157
        _timeout = 2
158 159 160
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.sock.settimeout(_timeout)
        self.sock.bind(self.addr_local)
161 162

        _t1 = time.time()
163
        self.failUnlessRaises(socket.error, self.sock.recvfrom, 8192)
164 165 166
        _t2 = time.time()

        _delta = abs(_t1 - _t2)
167 168 169
        self.assert_(_delta < _timeout + self.fuzz,
                     "timeout (%g) is %g seconds more than expected (%g)"
                     %(_delta, self.fuzz, _timeout))
170 171

    def testSend(self):
172
        # Test send() timeout
173 174 175 176
        # couldn't figure out how to test it
        pass

    def testSendto(self):
177
        # Test sendto() timeout
178 179 180 181
        # couldn't figure out how to test it
        pass

    def testSendall(self):
182
        # Test sendall() timeout
183 184 185 186
        # couldn't figure out how to test it
        pass


187
def test_main():
188
    test_support.requires('network')
189
    test_support.run_unittest(CreationTestCase, TimeoutTestCase)
190 191

if __name__ == "__main__":
192
    test_main()