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.addr_local  = ('127.0.0.1', 25339)
 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.failUnlessRaises(socket.error, self.sock.connect, addr)
 123         _t2 = time.time()
 124
 125         _delta = abs(_t1 - _t2)
 126         self.assert_(_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.failUnlessRaises(socket.error, self.sock.recv, 1024)
 138         _t2 = time.time()
 139
 140         _delta = abs(_t1 - _t2)
 141         self.assert_(_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         self.sock.bind(self.addr_local)
 150         self.sock.listen(5)
 151
 152         _t1 = time.time()
 153         self.failUnlessRaises(socket.error, self.sock.accept)
 154         _t2 = time.time()
 155
 156         _delta = abs(_t1 - _t2)
 157         self.assert_(_delta < _timeout + self.fuzz,
 158                      "timeout (%g) is %g seconds more than expected (%g)"
 159                      %(_delta, self.fuzz, _timeout))
 160
 161     def testRecvfromTimeout(self):
 162         # Test recvfrom() timeout
 163         _timeout = 2
 164         self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
 165         self.sock.settimeout(_timeout)
 166         self.sock.bind(self.addr_local)
 167
 168         _t1 = time.time()
 169         self.failUnlessRaises(socket.error, self.sock.recvfrom, 8192)
 170         _t2 = time.time()
 171
 172         _delta = abs(_t1 - _t2)
 173         self.assert_(_delta < _timeout + self.fuzz,
 174                      "timeout (%g) is %g seconds more than expected (%g)"
 175                      %(_delta, self.fuzz, _timeout))
 176
 177     def testSend(self):
 178         # Test send() timeout
 179         # couldn't figure out how to test it
 180         pass
 181
 182     def testSendto(self):
 183         # Test sendto() timeout
 184         # couldn't figure out how to test it
 185         pass
 186
 187     def testSendall(self):
 188         # Test sendall() timeout
 189         # couldn't figure out how to test it
 190         pass
 191
 192
 193 def test_main():
 194     test_support.requires('network')
 195     test_support.run_unittest(CreationTestCase, TimeoutTestCase)
 196
 197 if __name__ == "__main__":
 198     test_main()