"""Unit tests for socket timeout feature."""
from test
import test_support
# This requires the 'network' resource as given on the regrtest command line.
skip_expected
= not test_support
.is_resource_enabled('network')
class CreationTestCase(unittest
.TestCase
):
"""Test case for socket.gettimeout() and socket.settimeout()"""
self
.sock
= socket
.socket(socket
.AF_INET
, socket
.SOCK_STREAM
)
def testObjectCreation(self
):
self
.assertEqual(self
.sock
.gettimeout(), None,
"timeout not disabled by default")
def testFloatReturnValue(self
):
# Test return value of gettimeout()
self
.sock
.settimeout(7.345)
self
.assertEqual(self
.sock
.gettimeout(), 7.345)
self
.assertEqual(self
.sock
.gettimeout(), 3)
self
.sock
.settimeout(None)
self
.assertEqual(self
.sock
.gettimeout(), None)
def testReturnType(self
):
# Test return type of gettimeout()
self
.assertEqual(type(self
.sock
.gettimeout()), type(1.0))
self
.sock
.settimeout(3.9)
self
.assertEqual(type(self
.sock
.gettimeout()), type(1.0))
# Test type checking by settimeout()
self
.sock
.settimeout(0.0)
self
.sock
.settimeout(None)
self
.assertRaises(TypeError, self
.sock
.settimeout
, "")
self
.assertRaises(TypeError, self
.sock
.settimeout
, u
"")
self
.assertRaises(TypeError, self
.sock
.settimeout
, ())
self
.assertRaises(TypeError, self
.sock
.settimeout
, [])
self
.assertRaises(TypeError, self
.sock
.settimeout
, {})
self
.assertRaises(TypeError, self
.sock
.settimeout
, 0j
)
def testRangeCheck(self
):
# Test range checking by settimeout()
self
.assertRaises(ValueError, self
.sock
.settimeout
, -1)
self
.assertRaises(ValueError, self
.sock
.settimeout
, -1L)
self
.assertRaises(ValueError, self
.sock
.settimeout
, -1.0)
def testTimeoutThenBlocking(self
):
# Test settimeout() followed by setblocking()
self
.assertEqual(self
.sock
.gettimeout(), None)
self
.assertEqual(self
.sock
.gettimeout(), 0.0)
self
.assertEqual(self
.sock
.gettimeout(), 0.0)
self
.assertEqual(self
.sock
.gettimeout(), None)
def testBlockingThenTimeout(self
):
# Test setblocking() followed by settimeout()
self
.assertEqual(self
.sock
.gettimeout(), 1)
self
.assertEqual(self
.sock
.gettimeout(), 1)
class TimeoutTestCase(unittest
.TestCase
):
"""Test case for socket.socket() timeout functions"""
# There are a number of tests here trying to make sure that an operation
# doesn't take too much longer than expected. But competing machine
# activity makes it inevitable that such tests will fail at times.
# When fuzz was at 1.0, I (tim) routinely saw bogus failures on Win2K
# and Win98SE. Boosting it to 2.0 helped a lot, but isn't a real
self
.sock
= socket
.socket(socket
.AF_INET
, socket
.SOCK_STREAM
)
self
.addr_remote
= ('www.python.org', 80)
self
.addr_local
= ('127.0.0.1', 25339)
def testConnectTimeout(self
):
self
.sock
.settimeout(_timeout
)
self
.failUnlessRaises(socket
.error
, self
.sock
.connect
,
self
.assert_(_delta
< _timeout
+ self
.fuzz
,
"timeout (%g) is more than %g seconds more than expected (%g)"
%(_delta
, self
.fuzz
, _timeout
))
def testRecvTimeout(self
):
self
.sock
.connect(self
.addr_remote
)
self
.sock
.settimeout(_timeout
)
self
.failUnlessRaises(socket
.error
, self
.sock
.recv
, 1024)
self
.assert_(_delta
< _timeout
+ self
.fuzz
,
"timeout (%g) is %g seconds more than expected (%g)"
%(_delta
, self
.fuzz
, _timeout
))
def testAcceptTimeout(self
):
self
.sock
.settimeout(_timeout
)
self
.sock
.bind(self
.addr_local
)
self
.failUnlessRaises(socket
.error
, self
.sock
.accept
)
self
.assert_(_delta
< _timeout
+ self
.fuzz
,
"timeout (%g) is %g seconds more than expected (%g)"
%(_delta
, self
.fuzz
, _timeout
))
def testRecvfromTimeout(self
):
# Test recvfrom() timeout
self
.sock
= socket
.socket(socket
.AF_INET
, socket
.SOCK_DGRAM
)
self
.sock
.settimeout(_timeout
)
self
.sock
.bind(self
.addr_local
)
self
.failUnlessRaises(socket
.error
, self
.sock
.recvfrom
, 8192)
self
.assert_(_delta
< _timeout
+ self
.fuzz
,
"timeout (%g) is %g seconds more than expected (%g)"
%(_delta
, self
.fuzz
, _timeout
))
# couldn't figure out how to test it
# couldn't figure out how to test it
# couldn't figure out how to test it
test_support
.requires('network')
test_support
.run_unittest(CreationTestCase
, TimeoutTestCase
)
if __name__
== "__main__":