1"""Unit tests for socket timeout feature."""
2
3import unittest
4from test import test_support
5
6# This requires the 'network' resource as given on the regrtest command line.
7skip_expected = not test_support.is_resource_enabled('network')
8
9import time
10import socket
11
12
13class 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        self.assertEqual(self.sock.gettimeout(), None,
25                         "timeout not disabled by default")
26
27    def testFloatReturnValue(self):
28        # Test return value of gettimeout()
29        self.sock.settimeout(7.345)
30        self.assertEqual(self.sock.gettimeout(), 7.345)
31
32        self.sock.settimeout(3)
33        self.assertEqual(self.sock.gettimeout(), 3)
34
35        self.sock.settimeout(None)
36        self.assertEqual(self.sock.gettimeout(), None)
37
38    def testReturnType(self):
39        # Test return type of gettimeout()
40        self.sock.settimeout(1)
41        self.assertEqual(type(self.sock.gettimeout()), type(1.0))
42
43        self.sock.settimeout(3.9)
44        self.assertEqual(type(self.sock.gettimeout()), type(1.0))
45
46    def testTypeCheck(self):
47        # Test type checking by settimeout()
48        self.sock.settimeout(0)
49        self.sock.settimeout(0L)
50        self.sock.settimeout(0.0)
51        self.sock.settimeout(None)
52        self.assertRaises(TypeError, self.sock.settimeout, "")
53        self.assertRaises(TypeError, self.sock.settimeout, u"")
54        self.assertRaises(TypeError, self.sock.settimeout, ())
55        self.assertRaises(TypeError, self.sock.settimeout, [])
56        self.assertRaises(TypeError, self.sock.settimeout, {})
57        self.assertRaises(TypeError, self.sock.settimeout, 0j)
58
59    def testRangeCheck(self):
60        # Test range checking by settimeout()
61        self.assertRaises(ValueError, self.sock.settimeout, -1)
62        self.assertRaises(ValueError, self.sock.settimeout, -1L)
63        self.assertRaises(ValueError, self.sock.settimeout, -1.0)
64
65    def testTimeoutThenBlocking(self):
66        # Test settimeout() followed by setblocking()
67        self.sock.settimeout(10)
68        self.sock.setblocking(1)
69        self.assertEqual(self.sock.gettimeout(), None)
70        self.sock.setblocking(0)
71        self.assertEqual(self.sock.gettimeout(), 0.0)
72
73        self.sock.settimeout(10)
74        self.sock.setblocking(0)
75        self.assertEqual(self.sock.gettimeout(), 0.0)
76        self.sock.setblocking(1)
77        self.assertEqual(self.sock.gettimeout(), None)
78
79    def testBlockingThenTimeout(self):
80        # Test setblocking() followed by settimeout()
81        self.sock.setblocking(0)
82        self.sock.settimeout(1)
83        self.assertEqual(self.sock.gettimeout(), 1)
84
85        self.sock.setblocking(1)
86        self.sock.settimeout(1)
87        self.assertEqual(self.sock.gettimeout(), 1)
88
89
90class TimeoutTestCase(unittest.TestCase):
91    """Test case for socket.socket() timeout functions"""
92
93    # There are a number of tests here trying to make sure that an operation
94    # doesn't take too much longer than expected.  But competing machine
95    # activity makes it inevitable that such tests will fail at times.
96    # When fuzz was at 1.0, I (tim) routinely saw bogus failures on Win2K
97    # and Win98SE.  Boosting it to 2.0 helped a lot, but isn't a real
98    # solution.
99    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.localhost = '127.0.0.1'
105
106    def tearDown(self):
107        self.sock.close()
108
109    def testConnectTimeout(self):
110        # Choose a private address that is unlikely to exist to prevent
111        # failures due to the connect succeeding before the timeout.
112        # Use a dotted IP address to avoid including the DNS lookup time
113        # with the connect time.  This avoids failing the assertion that
114        # the timeout occurred fast enough.
115        addr = ('10.0.0.0', 12345)
116
117        # Test connect() timeout
118        _timeout = 0.001
119        self.sock.settimeout(_timeout)
120
121        _t1 = time.time()
122        self.assertRaises(socket.error, self.sock.connect, addr)
123        _t2 = time.time()
124
125        _delta = abs(_t1 - _t2)
126        self.assertTrue(_delta < _timeout + self.fuzz,
127                     "timeout (%g) is more than %g seconds more than expected (%g)"
128                     %(_delta, self.fuzz, _timeout))
129
130    def testRecvTimeout(self):
131        # Test recv() timeout
132        _timeout = 0.02
133
134        with test_support.transient_internet(self.addr_remote[0]):
135            self.sock.connect(self.addr_remote)
136            self.sock.settimeout(_timeout)
137
138            _t1 = time.time()
139            self.assertRaises(socket.timeout, self.sock.recv, 1024)
140            _t2 = time.time()
141
142            _delta = abs(_t1 - _t2)
143            self.assertTrue(_delta < _timeout + self.fuzz,
144                         "timeout (%g) is %g seconds more than expected (%g)"
145                         %(_delta, self.fuzz, _timeout))
146
147    def testAcceptTimeout(self):
148        # Test accept() timeout
149        _timeout = 2
150        self.sock.settimeout(_timeout)
151        # Prevent "Address already in use" socket exceptions
152        test_support.bind_port(self.sock, self.localhost)
153        self.sock.listen(5)
154
155        _t1 = time.time()
156        self.assertRaises(socket.error, self.sock.accept)
157        _t2 = time.time()
158
159        _delta = abs(_t1 - _t2)
160        self.assertTrue(_delta < _timeout + self.fuzz,
161                     "timeout (%g) is %g seconds more than expected (%g)"
162                     %(_delta, self.fuzz, _timeout))
163
164    def testRecvfromTimeout(self):
165        # Test recvfrom() timeout
166        _timeout = 2
167        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
168        self.sock.settimeout(_timeout)
169        # Prevent "Address already in use" socket exceptions
170        test_support.bind_port(self.sock, self.localhost)
171
172        _t1 = time.time()
173        self.assertRaises(socket.error, self.sock.recvfrom, 8192)
174        _t2 = time.time()
175
176        _delta = abs(_t1 - _t2)
177        self.assertTrue(_delta < _timeout + self.fuzz,
178                     "timeout (%g) is %g seconds more than expected (%g)"
179                     %(_delta, self.fuzz, _timeout))
180
181    def testSend(self):
182        # Test send() timeout
183        # couldn't figure out how to test it
184        pass
185
186    def testSendto(self):
187        # Test sendto() timeout
188        # couldn't figure out how to test it
189        pass
190
191    def testSendall(self):
192        # Test sendall() timeout
193        # couldn't figure out how to test it
194        pass
195
196
197def test_main():
198    test_support.requires('network')
199    test_support.run_unittest(CreationTestCase, TimeoutTestCase)
200
201if __name__ == "__main__":
202    test_main()
203