+++ /dev/null
-"""Unit tests for socket timeout feature."""\r
-\r
-import unittest\r
-from test import test_support\r
-\r
-# This requires the 'network' resource as given on the regrtest command line.\r
-skip_expected = not test_support.is_resource_enabled('network')\r
-\r
-import time\r
-import socket\r
-\r
-\r
-class CreationTestCase(unittest.TestCase):\r
- """Test case for socket.gettimeout() and socket.settimeout()"""\r
-\r
- def setUp(self):\r
- self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)\r
-\r
- def tearDown(self):\r
- self.sock.close()\r
-\r
- def testObjectCreation(self):\r
- # Test Socket creation\r
- self.assertEqual(self.sock.gettimeout(), None,\r
- "timeout not disabled by default")\r
-\r
- def testFloatReturnValue(self):\r
- # Test return value of gettimeout()\r
- self.sock.settimeout(7.345)\r
- self.assertEqual(self.sock.gettimeout(), 7.345)\r
-\r
- self.sock.settimeout(3)\r
- self.assertEqual(self.sock.gettimeout(), 3)\r
-\r
- self.sock.settimeout(None)\r
- self.assertEqual(self.sock.gettimeout(), None)\r
-\r
- def testReturnType(self):\r
- # Test return type of gettimeout()\r
- self.sock.settimeout(1)\r
- self.assertEqual(type(self.sock.gettimeout()), type(1.0))\r
-\r
- self.sock.settimeout(3.9)\r
- self.assertEqual(type(self.sock.gettimeout()), type(1.0))\r
-\r
- def testTypeCheck(self):\r
- # Test type checking by settimeout()\r
- self.sock.settimeout(0)\r
- self.sock.settimeout(0L)\r
- self.sock.settimeout(0.0)\r
- self.sock.settimeout(None)\r
- self.assertRaises(TypeError, self.sock.settimeout, "")\r
- self.assertRaises(TypeError, self.sock.settimeout, u"")\r
- self.assertRaises(TypeError, self.sock.settimeout, ())\r
- self.assertRaises(TypeError, self.sock.settimeout, [])\r
- self.assertRaises(TypeError, self.sock.settimeout, {})\r
- self.assertRaises(TypeError, self.sock.settimeout, 0j)\r
-\r
- def testRangeCheck(self):\r
- # Test range checking by settimeout()\r
- self.assertRaises(ValueError, self.sock.settimeout, -1)\r
- self.assertRaises(ValueError, self.sock.settimeout, -1L)\r
- self.assertRaises(ValueError, self.sock.settimeout, -1.0)\r
-\r
- def testTimeoutThenBlocking(self):\r
- # Test settimeout() followed by setblocking()\r
- self.sock.settimeout(10)\r
- self.sock.setblocking(1)\r
- self.assertEqual(self.sock.gettimeout(), None)\r
- self.sock.setblocking(0)\r
- self.assertEqual(self.sock.gettimeout(), 0.0)\r
-\r
- self.sock.settimeout(10)\r
- self.sock.setblocking(0)\r
- self.assertEqual(self.sock.gettimeout(), 0.0)\r
- self.sock.setblocking(1)\r
- self.assertEqual(self.sock.gettimeout(), None)\r
-\r
- def testBlockingThenTimeout(self):\r
- # Test setblocking() followed by settimeout()\r
- self.sock.setblocking(0)\r
- self.sock.settimeout(1)\r
- self.assertEqual(self.sock.gettimeout(), 1)\r
-\r
- self.sock.setblocking(1)\r
- self.sock.settimeout(1)\r
- self.assertEqual(self.sock.gettimeout(), 1)\r
-\r
-\r
-class TimeoutTestCase(unittest.TestCase):\r
- """Test case for socket.socket() timeout functions"""\r
-\r
- # There are a number of tests here trying to make sure that an operation\r
- # doesn't take too much longer than expected. But competing machine\r
- # activity makes it inevitable that such tests will fail at times.\r
- # When fuzz was at 1.0, I (tim) routinely saw bogus failures on Win2K\r
- # and Win98SE. Boosting it to 2.0 helped a lot, but isn't a real\r
- # solution.\r
- fuzz = 2.0\r
-\r
- def setUp(self):\r
- self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)\r
- self.addr_remote = ('www.python.org.', 80)\r
- self.localhost = '127.0.0.1'\r
-\r
- def tearDown(self):\r
- self.sock.close()\r
-\r
- def testConnectTimeout(self):\r
- # Choose a private address that is unlikely to exist to prevent\r
- # failures due to the connect succeeding before the timeout.\r
- # Use a dotted IP address to avoid including the DNS lookup time\r
- # with the connect time. This avoids failing the assertion that\r
- # the timeout occurred fast enough.\r
- addr = ('10.0.0.0', 12345)\r
-\r
- # Test connect() timeout\r
- _timeout = 0.001\r
- self.sock.settimeout(_timeout)\r
-\r
- _t1 = time.time()\r
- self.assertRaises(socket.error, self.sock.connect, addr)\r
- _t2 = time.time()\r
-\r
- _delta = abs(_t1 - _t2)\r
- self.assertTrue(_delta < _timeout + self.fuzz,\r
- "timeout (%g) is more than %g seconds more than expected (%g)"\r
- %(_delta, self.fuzz, _timeout))\r
-\r
- def testRecvTimeout(self):\r
- # Test recv() timeout\r
- _timeout = 0.02\r
-\r
- with test_support.transient_internet(self.addr_remote[0]):\r
- self.sock.connect(self.addr_remote)\r
- self.sock.settimeout(_timeout)\r
-\r
- _t1 = time.time()\r
- self.assertRaises(socket.timeout, self.sock.recv, 1024)\r
- _t2 = time.time()\r
-\r
- _delta = abs(_t1 - _t2)\r
- self.assertTrue(_delta < _timeout + self.fuzz,\r
- "timeout (%g) is %g seconds more than expected (%g)"\r
- %(_delta, self.fuzz, _timeout))\r
-\r
- def testAcceptTimeout(self):\r
- # Test accept() timeout\r
- _timeout = 2\r
- self.sock.settimeout(_timeout)\r
- # Prevent "Address already in use" socket exceptions\r
- test_support.bind_port(self.sock, self.localhost)\r
- self.sock.listen(5)\r
-\r
- _t1 = time.time()\r
- self.assertRaises(socket.error, self.sock.accept)\r
- _t2 = time.time()\r
-\r
- _delta = abs(_t1 - _t2)\r
- self.assertTrue(_delta < _timeout + self.fuzz,\r
- "timeout (%g) is %g seconds more than expected (%g)"\r
- %(_delta, self.fuzz, _timeout))\r
-\r
- def testRecvfromTimeout(self):\r
- # Test recvfrom() timeout\r
- _timeout = 2\r
- self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)\r
- self.sock.settimeout(_timeout)\r
- # Prevent "Address already in use" socket exceptions\r
- test_support.bind_port(self.sock, self.localhost)\r
-\r
- _t1 = time.time()\r
- self.assertRaises(socket.error, self.sock.recvfrom, 8192)\r
- _t2 = time.time()\r
-\r
- _delta = abs(_t1 - _t2)\r
- self.assertTrue(_delta < _timeout + self.fuzz,\r
- "timeout (%g) is %g seconds more than expected (%g)"\r
- %(_delta, self.fuzz, _timeout))\r
-\r
- def testSend(self):\r
- # Test send() timeout\r
- # couldn't figure out how to test it\r
- pass\r
-\r
- def testSendto(self):\r
- # Test sendto() timeout\r
- # couldn't figure out how to test it\r
- pass\r
-\r
- def testSendall(self):\r
- # Test sendall() timeout\r
- # couldn't figure out how to test it\r
- pass\r
-\r
-\r
-def test_main():\r
- test_support.requires('network')\r
- test_support.run_unittest(CreationTestCase, TimeoutTestCase)\r
-\r
-if __name__ == "__main__":\r
- test_main()\r
projects / mirror_edk2.git / blobdiff