PyXR

c:\python24\lib \ test \ test_timeout.py



0001 """Unit tests for socket timeout feature."""
0002 
0003 import unittest
0004 from test import test_support
0005 
0006 # This requires the 'network' resource as given on the regrtest command line.
0007 skip_expected = not test_support.is_resource_enabled('network')
0008 
0009 import time
0010 import socket
0011 
0012 
0013 class CreationTestCase(unittest.TestCase):
0014     """Test case for socket.gettimeout() and socket.settimeout()"""
0015 
0016     def setUp(self):
0017         self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
0018 
0019     def tearDown(self):
0020         self.sock.close()
0021 
0022     def testObjectCreation(self):
0023         # Test Socket creation
0024         self.assertEqual(self.sock.gettimeout(), None,
0025                          "timeout not disabled by default")
0026 
0027     def testFloatReturnValue(self):
0028         # Test return value of gettimeout()
0029         self.sock.settimeout(7.345)
0030         self.assertEqual(self.sock.gettimeout(), 7.345)
0031 
0032         self.sock.settimeout(3)
0033         self.assertEqual(self.sock.gettimeout(), 3)
0034 
0035         self.sock.settimeout(None)
0036         self.assertEqual(self.sock.gettimeout(), None)
0037 
0038     def testReturnType(self):
0039         # Test return type of gettimeout()
0040         self.sock.settimeout(1)
0041         self.assertEqual(type(self.sock.gettimeout()), type(1.0))
0042 
0043         self.sock.settimeout(3.9)
0044         self.assertEqual(type(self.sock.gettimeout()), type(1.0))
0045 
0046     def testTypeCheck(self):
0047         # Test type checking by settimeout()
0048         self.sock.settimeout(0)
0049         self.sock.settimeout(0L)
0050         self.sock.settimeout(0.0)
0051         self.sock.settimeout(None)
0052         self.assertRaises(TypeError, self.sock.settimeout, "")
0053         self.assertRaises(TypeError, self.sock.settimeout, u"")
0054         self.assertRaises(TypeError, self.sock.settimeout, ())
0055         self.assertRaises(TypeError, self.sock.settimeout, [])
0056         self.assertRaises(TypeError, self.sock.settimeout, {})
0057         self.assertRaises(TypeError, self.sock.settimeout, 0j)
0058 
0059     def testRangeCheck(self):
0060         # Test range checking by settimeout()
0061         self.assertRaises(ValueError, self.sock.settimeout, -1)
0062         self.assertRaises(ValueError, self.sock.settimeout, -1L)
0063         self.assertRaises(ValueError, self.sock.settimeout, -1.0)
0064 
0065     def testTimeoutThenBlocking(self):
0066         # Test settimeout() followed by setblocking()
0067         self.sock.settimeout(10)
0068         self.sock.setblocking(1)
0069         self.assertEqual(self.sock.gettimeout(), None)
0070         self.sock.setblocking(0)
0071         self.assertEqual(self.sock.gettimeout(), 0.0)
0072 
0073         self.sock.settimeout(10)
0074         self.sock.setblocking(0)
0075         self.assertEqual(self.sock.gettimeout(), 0.0)
0076         self.sock.setblocking(1)
0077         self.assertEqual(self.sock.gettimeout(), None)
0078 
0079     def testBlockingThenTimeout(self):
0080         # Test setblocking() followed by settimeout()
0081         self.sock.setblocking(0)
0082         self.sock.settimeout(1)
0083         self.assertEqual(self.sock.gettimeout(), 1)
0084 
0085         self.sock.setblocking(1)
0086         self.sock.settimeout(1)
0087         self.assertEqual(self.sock.gettimeout(), 1)
0088 
0089 
0090 class TimeoutTestCase(unittest.TestCase):
0091     """Test case for socket.socket() timeout functions"""
0092 
0093     # There are a number of tests here trying to make sure that an operation
0094     # doesn't take too much longer than expected.  But competing machine
0095     # activity makes it inevitable that such tests will fail at times.
0096     # When fuzz was at 1.0, I (tim) routinely saw bogus failures on Win2K
0097     # and Win98SE.  Boosting it to 2.0 helped a lot, but isn't a real
0098     # solution.
0099     fuzz = 2.0
0100 
0101     def setUp(self):
0102         self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
0103         self.addr_remote = ('www.python.org', 80)
0104         self.addr_local  = ('127.0.0.1', 25339)
0105 
0106     def tearDown(self):
0107         self.sock.close()
0108 
0109     def testConnectTimeout(self):
0110         # Test connect() timeout
0111         _timeout = 0.001
0112         self.sock.settimeout(_timeout)
0113 
0114         _t1 = time.time()
0115         self.failUnlessRaises(socket.error, self.sock.connect,
0116                 self.addr_remote)
0117         _t2 = time.time()
0118 
0119         _delta = abs(_t1 - _t2)
0120         self.assert_(_delta < _timeout + self.fuzz,
0121                      "timeout (%g) is more than %g seconds more than expected (%g)"
0122                      %(_delta, self.fuzz, _timeout))
0123 
0124     def testRecvTimeout(self):
0125         # Test recv() timeout
0126         _timeout = 0.02
0127         self.sock.connect(self.addr_remote)
0128         self.sock.settimeout(_timeout)
0129 
0130         _t1 = time.time()
0131         self.failUnlessRaises(socket.error, self.sock.recv, 1024)
0132         _t2 = time.time()
0133 
0134         _delta = abs(_t1 - _t2)
0135         self.assert_(_delta < _timeout + self.fuzz,
0136                      "timeout (%g) is %g seconds more than expected (%g)"
0137                      %(_delta, self.fuzz, _timeout))
0138 
0139     def testAcceptTimeout(self):
0140         # Test accept() timeout
0141         _timeout = 2
0142         self.sock.settimeout(_timeout)
0143         self.sock.bind(self.addr_local)
0144         self.sock.listen(5)
0145 
0146         _t1 = time.time()
0147         self.failUnlessRaises(socket.error, self.sock.accept)
0148         _t2 = time.time()
0149 
0150         _delta = abs(_t1 - _t2)
0151         self.assert_(_delta < _timeout + self.fuzz,
0152                      "timeout (%g) is %g seconds more than expected (%g)"
0153                      %(_delta, self.fuzz, _timeout))
0154 
0155     def testRecvfromTimeout(self):
0156         # Test recvfrom() timeout
0157         _timeout = 2
0158         self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
0159         self.sock.settimeout(_timeout)
0160         self.sock.bind(self.addr_local)
0161 
0162         _t1 = time.time()
0163         self.failUnlessRaises(socket.error, self.sock.recvfrom, 8192)
0164         _t2 = time.time()
0165 
0166         _delta = abs(_t1 - _t2)
0167         self.assert_(_delta < _timeout + self.fuzz,
0168                      "timeout (%g) is %g seconds more than expected (%g)"
0169                      %(_delta, self.fuzz, _timeout))
0170 
0171     def testSend(self):
0172         # Test send() timeout
0173         # couldn't figure out how to test it
0174         pass
0175 
0176     def testSendto(self):
0177         # Test sendto() timeout
0178         # couldn't figure out how to test it
0179         pass
0180 
0181     def testSendall(self):
0182         # Test sendall() timeout
0183         # couldn't figure out how to test it
0184         pass
0185 
0186 
0187 def test_main():
0188     test_support.requires('network')
0189     test_support.run_unittest(CreationTestCase, TimeoutTestCase)
0190 
0191 if __name__ == "__main__":
0192     test_main()
0193 

Generated by PyXR 0.9.4
SourceForge.net Logo