Lib/test/test_timeout.py

   1 """Unit tests for socket timeout feature."""
   2
   3 import unittest
   4 from test import test_support
   5
   6 import time
   7 import socket
   8
   9
  10 class CreationTestCase(unittest.TestCase):
  11     """Test case for socket.gettimeout() and socket.settimeout()"""
  12
  13     def setUp(self):
  14         self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  15
  16     def tearDown(self):
  17         self.sock.close()
  18
  19     def testObjectCreation(self):
  20         # Test Socket creation
  21         self.assertEqual(self.sock.gettimeout(), None,
  22                          "timeout not disabled by default")
  23
  24     def testFloatReturnValue(self):
  25         # Test return value of gettimeout()
  26         self.sock.settimeout(7.345)
  27         self.assertEqual(self.sock.gettimeout(), 7.345)
  28
  29         self.sock.settimeout(3)
  30         self.assertEqual(self.sock.gettimeout(), 3)
  31
  32         self.sock.settimeout(None)
  33         self.assertEqual(self.sock.gettimeout(), None)
  34
  35     def testReturnType(self):
  36         # Test return type of gettimeout()
  37         self.sock.settimeout(1)
  38         self.assertEqual(type(self.sock.gettimeout()), type(1.0))
  39
  40         self.sock.settimeout(3.9)
  41         self.assertEqual(type(self.sock.gettimeout()), type(1.0))
  42
  43     def testTypeCheck(self):
  44         # Test type checking by settimeout()
  45         self.sock.settimeout(0)
  46         self.sock.settimeout(0L)
  47         self.sock.settimeout(0.0)
  48         self.sock.settimeout(None)
  49         self.assertRaises(TypeError, self.sock.settimeout, "")
  50         self.assertRaises(TypeError, self.sock.settimeout, u"")
  51         self.assertRaises(TypeError, self.sock.settimeout, ())
  52         self.assertRaises(TypeError, self.sock.settimeout, [])
  53         self.assertRaises(TypeError, self.sock.settimeout, {})
  54         self.assertRaises(TypeError, self.sock.settimeout, 0j)
  55
  56     def testRangeCheck(self):
  57         # Test range checking by settimeout()
  58         self.assertRaises(ValueError, self.sock.settimeout, -1)
  59         self.assertRaises(ValueError, self.sock.settimeout, -1L)
  60         self.assertRaises(ValueError, self.sock.settimeout, -1.0)
  61
  62     def testTimeoutThenBlocking(self):
  63         # Test settimeout() followed by setblocking()
  64         self.sock.settimeout(10)
  65         self.sock.setblocking(1)
  66         self.assertEqual(self.sock.gettimeout(), None)
  67         self.sock.setblocking(0)
  68         self.assertEqual(self.sock.gettimeout(), 0.0)
  69
  70         self.sock.settimeout(10)
  71         self.sock.setblocking(0)
  72         self.assertEqual(self.sock.gettimeout(), 0.0)
  73         self.sock.setblocking(1)
  74         self.assertEqual(self.sock.gettimeout(), None)
  75
  76     def testBlockingThenTimeout(self):
  77         # Test setblocking() followed by settimeout()
  78         self.sock.setblocking(0)
  79         self.sock.settimeout(1)
  80         self.assertEqual(self.sock.gettimeout(), 1)
  81
  82         self.sock.setblocking(1)
  83         self.sock.settimeout(1)
  84         self.assertEqual(self.sock.gettimeout(), 1)
  85
  86
  87 class TimeoutTestCase(unittest.TestCase):
  88     """Test case for socket.socket() timeout functions"""
  89
  90     fuzz = 1.0
  91
  92     def setUp(self):
  93         self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  94         self.addr_remote = ('www.google.com', 80)
  95         self.addr_local  = ('127.0.0.1', 25339)
  96
  97     def tearDown(self):
  98         self.sock.close()
  99
 100     def testConnectTimeout(self):
 101         # Test connect() timeout
 102         _timeout = 0.001
 103         self.sock.settimeout(_timeout)
 104
 105         _t1 = time.time()
 106         self.failUnlessRaises(socket.error, self.sock.connect,
 107                 self.addr_remote)
 108         _t2 = time.time()
 109
 110         _delta = abs(_t1 - _t2)
 111         self.assert_(_delta < _timeout + self.fuzz,
 112                      "timeout (%g) is %g seconds more than expected (%g)"
 113                      %(_delta, self.fuzz, _timeout))
 114
 115     def testRecvTimeout(self):
 116         # Test recv() timeout
 117         _timeout = 0.02
 118         self.sock.connect(self.addr_remote)
 119         self.sock.settimeout(_timeout)
 120
 121         _t1 = time.time()
 122         self.failUnlessRaises(socket.error, self.sock.recv, 1024)
 123         _t2 = time.time()
 124
 125         _delta = abs(_t1 - _t2)
 126         self.assert_(_delta < _timeout + self.fuzz,
 127                      "timeout (%g) is %g seconds more than expected (%g)"
 128                      %(_delta, self.fuzz, _timeout))
 129
 130     def testAcceptTimeout(self):
 131         # Test accept() timeout
 132         _timeout = 2
 133         self.sock.settimeout(_timeout)
 134         self.sock.bind(self.addr_local)
 135         self.sock.listen(5)
 136
 137         _t1 = time.time()
 138         self.failUnlessRaises(socket.error, self.sock.accept)
 139         _t2 = time.time()
 140
 141         _delta = abs(_t1 - _t2)
 142         self.assert_(_delta < _timeout + self.fuzz,
 143                      "timeout (%g) is %g seconds more than expected (%g)"
 144                      %(_delta, self.fuzz, _timeout))
 145
 146     def testRecvfromTimeout(self):
 147         # Test recvfrom() timeout
 148         _timeout = 2
 149         self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
 150         self.sock.settimeout(_timeout)
 151         self.sock.bind(self.addr_local)
 152
 153         _t1 = time.time()
 154         self.failUnlessRaises(socket.error, self.sock.recvfrom, 8192)
 155         _t2 = time.time()
 156
 157         _delta = abs(_t1 - _t2)
 158         self.assert_(_delta < _timeout + self.fuzz,
 159                      "timeout (%g) is %g seconds more than expected (%g)"
 160                      %(_delta, self.fuzz, _timeout))
 161
 162     def testSend(self):
 163         # Test send() timeout
 164         # couldn't figure out how to test it
 165         pass
 166
 167     def testSendto(self):
 168         # Test sendto() timeout
 169         # couldn't figure out how to test it
 170         pass
 171
 172     def testSendall(self):
 173         # Test sendall() timeout
 174         # couldn't figure out how to test it
 175         pass
 176
 177
 178 def main():
 179     suite = unittest.TestSuite()
 180     suite.addTest(unittest.makeSuite(CreationTestCase))
 181     suite.addTest(unittest.makeSuite(TimeoutTestCase))
 182     test_support.run_suite(suite)
 183
 184 if __name__ == "__main__":
 185     main()