Lib/test/test_timeout.py

   1 """Unit tests for socket timeout feature."""
   2
   3 import unittest
   4 from test import test_support
   5
   6 # This requires the 'network' resource as given on the regrtest command line.
   7 skip_expected = not test_support.is_resource_enabled('network')
   8
   9 import time
  10 import socket
  11
  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         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
  90 class 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         self.sock.connect(self.addr_remote)
 134         self.sock.settimeout(_timeout)
 135
 136         _t1 = time.time()
 137         self.assertRaises(socket.error, self.sock.recv, 1024)
 138         _t2 = time.time()
 139
 140         _delta = abs(_t1 - _t2)
 141         self.assertTrue(_delta < _timeout + self.fuzz,
 142                      "timeout (%g) is %g seconds more than expected (%g)"
 143                      %(_delta, self.fuzz, _timeout))
 144
 145     def testAcceptTimeout(self):
 146         # Test accept() timeout
 147         _timeout = 2
 148         self.sock.settimeout(_timeout)
 149         # Prevent "Address already in use" socket exceptions
 150         test_support.bind_port(self.sock, self.localhost)
 151         self.sock.listen(5)
 152
 153         _t1 = time.time()
 154         self.assertRaises(socket.error, self.sock.accept)
 155         _t2 = time.time()
 156
 157         _delta = abs(_t1 - _t2)
 158         self.assertTrue(_delta < _timeout + self.fuzz,
 159                      "timeout (%g) is %g seconds more than expected (%g)"
 160                      %(_delta, self.fuzz, _timeout))
 161
 162     def testRecvfromTimeout(self):
 163         # Test recvfrom() timeout
 164         _timeout = 2
 165         self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
 166         self.sock.settimeout(_timeout)
 167         # Prevent "Address already in use" socket exceptions
 168         test_support.bind_port(self.sock, self.localhost)
 169
 170         _t1 = time.time()
 171         self.assertRaises(socket.error, self.sock.recvfrom, 8192)
 172         _t2 = time.time()
 173
 174         _delta = abs(_t1 - _t2)
 175         self.assertTrue(_delta < _timeout + self.fuzz,
 176                      "timeout (%g) is %g seconds more than expected (%g)"
 177                      %(_delta, self.fuzz, _timeout))
 178
 179     def testSend(self):
 180         # Test send() timeout
 181         # couldn't figure out how to test it
 182         pass
 183
 184     def testSendto(self):
 185         # Test sendto() timeout
 186         # couldn't figure out how to test it
 187         pass
 188
 189     def testSendall(self):
 190         # Test sendall() timeout
 191         # couldn't figure out how to test it
 192         pass
 193
 194
 195 def test_main():
 196     test_support.requires('network')
 197     test_support.run_unittest(CreationTestCase, TimeoutTestCase)
 198
 199 if __name__ == "__main__":
 200     test_main()