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    @unittest.skip('test not implemented')
182    def testSend(self):
183        # Test send() timeout
184        # couldn't figure out how to test it
185        pass
186
187    @unittest.skip('test not implemented')
188    def testSendto(self):
189        # Test sendto() timeout
190        # couldn't figure out how to test it
191        pass
192
193    @unittest.skip('test not implemented')
194    def testSendall(self):
195        # Test sendall() timeout
196        # couldn't figure out how to test it
197        pass
198
199
200def test_main():
201    test_support.requires('network')
202    test_support.run_unittest(CreationTestCase, TimeoutTestCase)
203
204if __name__ == "__main__":
205    test_main()
206