| #!/usr/bin/env python |
| |
| import unittest |
| from test import test_support |
| |
| import socket |
| import select |
| import time |
| import thread, threading |
| import Queue |
| import sys |
| |
| PORT = 50007 |
| HOST = 'localhost' |
| MSG = 'Michael Gilfix was here\n' |
| |
| class SocketTCPTest(unittest.TestCase): |
| |
| def setUp(self): |
| self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| self.serv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) |
| self.serv.bind((HOST, PORT)) |
| self.serv.listen(1) |
| |
| def tearDown(self): |
| self.serv.close() |
| self.serv = None |
| |
| class SocketUDPTest(unittest.TestCase): |
| |
| def setUp(self): |
| self.serv = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) |
| self.serv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) |
| self.serv.bind((HOST, PORT)) |
| |
| def tearDown(self): |
| self.serv.close() |
| self.serv = None |
| |
| class ThreadableTest: |
| """Threadable Test class |
| |
| The ThreadableTest class makes it easy to create a threaded |
| client/server pair from an existing unit test. To create a |
| new threaded class from an existing unit test, use multiple |
| inheritance: |
| |
| class NewClass (OldClass, ThreadableTest): |
| pass |
| |
| This class defines two new fixture functions with obvious |
| purposes for overriding: |
| |
| clientSetUp () |
| clientTearDown () |
| |
| Any new test functions within the class must then define |
| tests in pairs, where the test name is preceeded with a |
| '_' to indicate the client portion of the test. Ex: |
| |
| def testFoo(self): |
| # Server portion |
| |
| def _testFoo(self): |
| # Client portion |
| |
| Any exceptions raised by the clients during their tests |
| are caught and transferred to the main thread to alert |
| the testing framework. |
| |
| Note, the server setup function cannot call any blocking |
| functions that rely on the client thread during setup, |
| unless serverExplicityReady() is called just before |
| the blocking call (such as in setting up a client/server |
| connection and performing the accept() in setUp(). |
| """ |
| |
| def __init__(self): |
| # Swap the true setup function |
| self.__setUp = self.setUp |
| self.__tearDown = self.tearDown |
| self.setUp = self._setUp |
| self.tearDown = self._tearDown |
| |
| def serverExplicitReady(self): |
| """This method allows the server to explicitly indicate that |
| it wants the client thread to proceed. This is useful if the |
| server is about to execute a blocking routine that is |
| dependent upon the client thread during its setup routine.""" |
| self.server_ready.set() |
| |
| def _setUp(self): |
| self.server_ready = threading.Event() |
| self.client_ready = threading.Event() |
| self.done = threading.Event() |
| self.queue = Queue.Queue(1) |
| |
| # Do some munging to start the client test. |
| methodname = self.id() |
| i = methodname.rfind('.') |
| methodname = methodname[i+1:] |
| test_method = getattr(self, '_' + methodname) |
| self.client_thread = thread.start_new_thread( |
| self.clientRun, (test_method,)) |
| |
| self.__setUp() |
| if not self.server_ready.isSet(): |
| self.server_ready.set() |
| self.client_ready.wait() |
| |
| def _tearDown(self): |
| self.__tearDown() |
| self.done.wait() |
| |
| if not self.queue.empty(): |
| msg = self.queue.get() |
| self.fail(msg) |
| |
| def clientRun(self, test_func): |
| self.server_ready.wait() |
| self.client_ready.set() |
| self.clientSetUp() |
| if not callable(test_func): |
| raise TypeError, "test_func must be a callable function" |
| try: |
| test_func() |
| except Exception, strerror: |
| self.queue.put(strerror) |
| self.clientTearDown() |
| |
| def clientSetUp(self): |
| raise NotImplementedError, "clientSetUp must be implemented." |
| |
| def clientTearDown(self): |
| self.done.set() |
| thread.exit() |
| |
| class ThreadedTCPSocketTest(SocketTCPTest, ThreadableTest): |
| |
| def __init__(self, methodName='runTest'): |
| SocketTCPTest.__init__(self, methodName=methodName) |
| ThreadableTest.__init__(self) |
| |
| def clientSetUp(self): |
| self.cli = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| |
| def clientTearDown(self): |
| self.cli.close() |
| self.cli = None |
| ThreadableTest.clientTearDown(self) |
| |
| class ThreadedUDPSocketTest(SocketUDPTest, ThreadableTest): |
| |
| def __init__(self, methodName='runTest'): |
| SocketUDPTest.__init__(self, methodName=methodName) |
| ThreadableTest.__init__(self) |
| |
| def clientSetUp(self): |
| self.cli = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) |
| |
| class SocketConnectedTest(ThreadedTCPSocketTest): |
| |
| def __init__(self, methodName='runTest'): |
| ThreadedTCPSocketTest.__init__(self, methodName=methodName) |
| |
| def setUp(self): |
| ThreadedTCPSocketTest.setUp(self) |
| # Indicate explicitly we're ready for the client thread to |
| # proceed and then perform the blocking call to accept |
| self.serverExplicitReady() |
| conn, addr = self.serv.accept() |
| self.cli_conn = conn |
| |
| def tearDown(self): |
| self.cli_conn.close() |
| self.cli_conn = None |
| ThreadedTCPSocketTest.tearDown(self) |
| |
| def clientSetUp(self): |
| ThreadedTCPSocketTest.clientSetUp(self) |
| self.cli.connect((HOST, PORT)) |
| self.serv_conn = self.cli |
| |
| def clientTearDown(self): |
| self.serv_conn.close() |
| self.serv_conn = None |
| ThreadedTCPSocketTest.clientTearDown(self) |
| |
| ####################################################################### |
| ## Begin Tests |
| |
| class GeneralModuleTests(unittest.TestCase): |
| |
| def testSocketError(self): |
| # Testing socket module exceptions |
| def raise_error(*args, **kwargs): |
| raise socket.error |
| def raise_herror(*args, **kwargs): |
| raise socket.herror |
| def raise_gaierror(*args, **kwargs): |
| raise socket.gaierror |
| self.failUnlessRaises(socket.error, raise_error, |
| "Error raising socket exception.") |
| self.failUnlessRaises(socket.error, raise_herror, |
| "Error raising socket exception.") |
| self.failUnlessRaises(socket.error, raise_gaierror, |
| "Error raising socket exception.") |
| |
| def testCrucialConstants(self): |
| # Testing for mission critical constants |
| socket.AF_INET |
| socket.SOCK_STREAM |
| socket.SOCK_DGRAM |
| socket.SOCK_RAW |
| socket.SOCK_RDM |
| socket.SOCK_SEQPACKET |
| socket.SOL_SOCKET |
| socket.SO_REUSEADDR |
| |
| def testHostnameRes(self): |
| # Testing hostname resolution mechanisms |
| hostname = socket.gethostname() |
| try: |
| ip = socket.gethostbyname(hostname) |
| except socket.error: |
| # Probably name lookup wasn't set up right; skip this test |
| return |
| self.assert_(ip.find('.') >= 0, "Error resolving host to ip.") |
| try: |
| hname, aliases, ipaddrs = socket.gethostbyaddr(ip) |
| except socket.error: |
| # Probably a similar problem as above; skip this test |
| return |
| all_host_names = [hname] + aliases |
| fqhn = socket.getfqdn() |
| if not fqhn in all_host_names: |
| self.fail("Error testing host resolution mechanisms.") |
| |
| def testRefCountGetNameInfo(self): |
| # Testing reference count for getnameinfo |
| import sys |
| if hasattr(sys, "getrefcount"): |
| try: |
| # On some versions, this loses a reference |
| orig = sys.getrefcount(__name__) |
| socket.getnameinfo(__name__,0) |
| except SystemError: |
| if sys.getrefcount(__name__) <> orig: |
| self.fail("socket.getnameinfo loses a reference") |
| |
| def testInterpreterCrash(self): |
| # Making sure getnameinfo doesn't crash the interpreter |
| try: |
| # On some versions, this crashes the interpreter. |
| socket.getnameinfo(('x', 0, 0, 0), 0) |
| except socket.error: |
| pass |
| |
| def testNtoH(self): |
| # This just checks that htons etc. are their own inverse, |
| # when looking at the lower 16 or 32 bits. |
| sizes = {socket.htonl: 32, socket.ntohl: 32, |
| socket.htons: 16, socket.ntohs: 16} |
| for func, size in sizes.items(): |
| mask = (1L<<size) - 1 |
| for i in (0, 1, 0xffff, ~0xffff, 2, 0x01234567, 0x76543210): |
| self.assertEqual(i & mask, func(func(i&mask)) & mask) |
| |
| swapped = func(mask) |
| self.assertEqual(swapped & mask, mask) |
| self.assertRaises(OverflowError, func, 1L<<34) |
| |
| def testGetServByName(self): |
| # Testing getservbyname() |
| # try a few protocols - not everyone has telnet enabled |
| found = 0 |
| for proto in ("telnet", "ssh", "www", "ftp"): |
| try: |
| socket.getservbyname(proto, 'tcp') |
| found = 1 |
| break |
| except socket.error: |
| pass |
| try: |
| socket.getservbyname(proto, 'udp') |
| found = 1 |
| break |
| except socket.error: |
| pass |
| if not found: |
| raise socket.error |
| |
| def testDefaultTimeout(self): |
| # Testing default timeout |
| # The default timeout should initially be None |
| self.assertEqual(socket.getdefaulttimeout(), None) |
| s = socket.socket() |
| self.assertEqual(s.gettimeout(), None) |
| s.close() |
| |
| # Set the default timeout to 10, and see if it propagates |
| socket.setdefaulttimeout(10) |
| self.assertEqual(socket.getdefaulttimeout(), 10) |
| s = socket.socket() |
| self.assertEqual(s.gettimeout(), 10) |
| s.close() |
| |
| # Reset the default timeout to None, and see if it propagates |
| socket.setdefaulttimeout(None) |
| self.assertEqual(socket.getdefaulttimeout(), None) |
| s = socket.socket() |
| self.assertEqual(s.gettimeout(), None) |
| s.close() |
| |
| # Check that setting it to an invalid value raises ValueError |
| self.assertRaises(ValueError, socket.setdefaulttimeout, -1) |
| |
| # Check that setting it to an invalid type raises TypeError |
| self.assertRaises(TypeError, socket.setdefaulttimeout, "spam") |
| |
| def testIPv4toString(self): |
| if not hasattr(socket, 'inet_pton'): |
| return # No inet_pton() on this platform |
| from socket import inet_aton as f, inet_pton, AF_INET |
| g = lambda a: inet_pton(AF_INET, a) |
| |
| self.assertEquals('\x00\x00\x00\x00', f('0.0.0.0')) |
| self.assertEquals('\xff\x00\xff\x00', f('255.0.255.0')) |
| self.assertEquals('\xaa\xaa\xaa\xaa', f('170.170.170.170')) |
| self.assertEquals('\x01\x02\x03\x04', f('1.2.3.4')) |
| |
| self.assertEquals('\x00\x00\x00\x00', g('0.0.0.0')) |
| self.assertEquals('\xff\x00\xff\x00', g('255.0.255.0')) |
| self.assertEquals('\xaa\xaa\xaa\xaa', g('170.170.170.170')) |
| |
| def testIPv6toString(self): |
| if not hasattr(socket, 'inet_pton'): |
| return # No inet_pton() on this platform |
| try: |
| from socket import inet_pton, AF_INET6, has_ipv6 |
| if not has_ipv6: |
| return |
| except ImportError: |
| return |
| f = lambda a: inet_pton(AF_INET6, a) |
| |
| self.assertEquals('\x00' * 16, f('::')) |
| self.assertEquals('\x00' * 16, f('0::0')) |
| self.assertEquals('\x00\x01' + '\x00' * 14, f('1::')) |
| self.assertEquals( |
| '\x45\xef\x76\xcb\x00\x1a\x56\xef\xaf\xeb\x0b\xac\x19\x24\xae\xae', |
| f('45ef:76cb:1a:56ef:afeb:bac:1924:aeae') |
| ) |
| |
| def testStringToIPv4(self): |
| if not hasattr(socket, 'inet_ntop'): |
| return # No inet_ntop() on this platform |
| from socket import inet_ntoa as f, inet_ntop, AF_INET |
| g = lambda a: inet_ntop(AF_INET, a) |
| |
| self.assertEquals('1.0.1.0', f('\x01\x00\x01\x00')) |
| self.assertEquals('170.85.170.85', f('\xaa\x55\xaa\x55')) |
| self.assertEquals('255.255.255.255', f('\xff\xff\xff\xff')) |
| self.assertEquals('1.2.3.4', f('\x01\x02\x03\x04')) |
| |
| self.assertEquals('1.0.1.0', g('\x01\x00\x01\x00')) |
| self.assertEquals('170.85.170.85', g('\xaa\x55\xaa\x55')) |
| self.assertEquals('255.255.255.255', g('\xff\xff\xff\xff')) |
| |
| def testStringToIPv6(self): |
| if not hasattr(socket, 'inet_ntop'): |
| return # No inet_ntop() on this platform |
| try: |
| from socket import inet_ntop, AF_INET6, has_ipv6 |
| if not has_ipv6: |
| return |
| except ImportError: |
| return |
| f = lambda a: inet_ntop(AF_INET6, a) |
| |
| self.assertEquals('::', f('\x00' * 16)) |
| self.assertEquals('::1', f('\x00' * 15 + '\x01')) |
| self.assertEquals( |
| 'aef:b01:506:1001:ffff:9997:55:170', |
| f('\x0a\xef\x0b\x01\x05\x06\x10\x01\xff\xff\x99\x97\x00\x55\x01\x70') |
| ) |
| |
| # XXX The following don't test module-level functionality... |
| |
| def testSockName(self): |
| # Testing getsockname() |
| sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| sock.bind(("0.0.0.0", PORT+1)) |
| name = sock.getsockname() |
| self.assertEqual(name, ("0.0.0.0", PORT+1)) |
| |
| def testGetSockOpt(self): |
| # Testing getsockopt() |
| # We know a socket should start without reuse==0 |
| sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| reuse = sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR) |
| self.failIf(reuse != 0, "initial mode is reuse") |
| |
| def testSetSockOpt(self): |
| # Testing setsockopt() |
| sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) |
| reuse = sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR) |
| self.failIf(reuse == 0, "failed to set reuse mode") |
| |
| def testSendAfterClose(self): |
| # testing send() after close() with timeout |
| sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| sock.settimeout(1) |
| sock.close() |
| self.assertRaises(socket.error, sock.send, "spam") |
| |
| class BasicTCPTest(SocketConnectedTest): |
| |
| def __init__(self, methodName='runTest'): |
| SocketConnectedTest.__init__(self, methodName=methodName) |
| |
| def testRecv(self): |
| # Testing large receive over TCP |
| msg = self.cli_conn.recv(1024) |
| self.assertEqual(msg, MSG) |
| |
| def _testRecv(self): |
| self.serv_conn.send(MSG) |
| |
| def testOverFlowRecv(self): |
| # Testing receive in chunks over TCP |
| seg1 = self.cli_conn.recv(len(MSG) - 3) |
| seg2 = self.cli_conn.recv(1024) |
| msg = seg1 + seg2 |
| self.assertEqual(msg, MSG) |
| |
| def _testOverFlowRecv(self): |
| self.serv_conn.send(MSG) |
| |
| def testRecvFrom(self): |
| # Testing large recvfrom() over TCP |
| msg, addr = self.cli_conn.recvfrom(1024) |
| self.assertEqual(msg, MSG) |
| |
| def _testRecvFrom(self): |
| self.serv_conn.send(MSG) |
| |
| def testOverFlowRecvFrom(self): |
| # Testing recvfrom() in chunks over TCP |
| seg1, addr = self.cli_conn.recvfrom(len(MSG)-3) |
| seg2, addr = self.cli_conn.recvfrom(1024) |
| msg = seg1 + seg2 |
| self.assertEqual(msg, MSG) |
| |
| def _testOverFlowRecvFrom(self): |
| self.serv_conn.send(MSG) |
| |
| def testSendAll(self): |
| # Testing sendall() with a 2048 byte string over TCP |
| msg = '' |
| while 1: |
| read = self.cli_conn.recv(1024) |
| if not read: |
| break |
| msg += read |
| self.assertEqual(msg, 'f' * 2048) |
| |
| def _testSendAll(self): |
| big_chunk = 'f' * 2048 |
| self.serv_conn.sendall(big_chunk) |
| |
| def testFromFd(self): |
| # Testing fromfd() |
| if not hasattr(socket, "fromfd"): |
| return # On Windows, this doesn't exist |
| fd = self.cli_conn.fileno() |
| sock = socket.fromfd(fd, socket.AF_INET, socket.SOCK_STREAM) |
| msg = sock.recv(1024) |
| self.assertEqual(msg, MSG) |
| |
| def _testFromFd(self): |
| self.serv_conn.send(MSG) |
| |
| def testShutdown(self): |
| # Testing shutdown() |
| msg = self.cli_conn.recv(1024) |
| self.assertEqual(msg, MSG) |
| |
| def _testShutdown(self): |
| self.serv_conn.send(MSG) |
| self.serv_conn.shutdown(2) |
| |
| class BasicUDPTest(ThreadedUDPSocketTest): |
| |
| def __init__(self, methodName='runTest'): |
| ThreadedUDPSocketTest.__init__(self, methodName=methodName) |
| |
| def testSendtoAndRecv(self): |
| # Testing sendto() and Recv() over UDP |
| msg = self.serv.recv(len(MSG)) |
| self.assertEqual(msg, MSG) |
| |
| def _testSendtoAndRecv(self): |
| self.cli.sendto(MSG, 0, (HOST, PORT)) |
| |
| def testRecvFrom(self): |
| # Testing recvfrom() over UDP |
| msg, addr = self.serv.recvfrom(len(MSG)) |
| self.assertEqual(msg, MSG) |
| |
| def _testRecvFrom(self): |
| self.cli.sendto(MSG, 0, (HOST, PORT)) |
| |
| class NonBlockingTCPTests(ThreadedTCPSocketTest): |
| |
| def __init__(self, methodName='runTest'): |
| ThreadedTCPSocketTest.__init__(self, methodName=methodName) |
| |
| def testSetBlocking(self): |
| # Testing whether set blocking works |
| self.serv.setblocking(0) |
| start = time.time() |
| try: |
| self.serv.accept() |
| except socket.error: |
| pass |
| end = time.time() |
| self.assert_((end - start) < 1.0, "Error setting non-blocking mode.") |
| |
| def _testSetBlocking(self): |
| pass |
| |
| def testAccept(self): |
| # Testing non-blocking accept |
| self.serv.setblocking(0) |
| try: |
| conn, addr = self.serv.accept() |
| except socket.error: |
| pass |
| else: |
| self.fail("Error trying to do non-blocking accept.") |
| read, write, err = select.select([self.serv], [], []) |
| if self.serv in read: |
| conn, addr = self.serv.accept() |
| else: |
| self.fail("Error trying to do accept after select.") |
| |
| def _testAccept(self): |
| time.sleep(0.1) |
| self.cli.connect((HOST, PORT)) |
| |
| def testConnect(self): |
| # Testing non-blocking connect |
| conn, addr = self.serv.accept() |
| |
| def _testConnect(self): |
| self.cli.settimeout(10) |
| self.cli.connect((HOST, PORT)) |
| |
| def testRecv(self): |
| # Testing non-blocking recv |
| conn, addr = self.serv.accept() |
| conn.setblocking(0) |
| try: |
| msg = conn.recv(len(MSG)) |
| except socket.error: |
| pass |
| else: |
| self.fail("Error trying to do non-blocking recv.") |
| read, write, err = select.select([conn], [], []) |
| if conn in read: |
| msg = conn.recv(len(MSG)) |
| self.assertEqual(msg, MSG) |
| else: |
| self.fail("Error during select call to non-blocking socket.") |
| |
| def _testRecv(self): |
| self.cli.connect((HOST, PORT)) |
| time.sleep(0.1) |
| self.cli.send(MSG) |
| |
| class FileObjectClassTestCase(SocketConnectedTest): |
| |
| bufsize = -1 # Use default buffer size |
| |
| def __init__(self, methodName='runTest'): |
| SocketConnectedTest.__init__(self, methodName=methodName) |
| |
| def setUp(self): |
| SocketConnectedTest.setUp(self) |
| self.serv_file = self.cli_conn.makefile('rb', self.bufsize) |
| |
| def tearDown(self): |
| self.serv_file.close() |
| self.serv_file = None |
| SocketConnectedTest.tearDown(self) |
| |
| def clientSetUp(self): |
| SocketConnectedTest.clientSetUp(self) |
| self.cli_file = self.serv_conn.makefile('wb') |
| |
| def clientTearDown(self): |
| self.cli_file.close() |
| self.cli_file = None |
| SocketConnectedTest.clientTearDown(self) |
| |
| def testSmallRead(self): |
| # Performing small file read test |
| first_seg = self.serv_file.read(len(MSG)-3) |
| second_seg = self.serv_file.read(3) |
| msg = first_seg + second_seg |
| self.assertEqual(msg, MSG) |
| |
| def _testSmallRead(self): |
| self.cli_file.write(MSG) |
| self.cli_file.flush() |
| |
| def testFullRead(self): |
| # read until EOF |
| msg = self.serv_file.read() |
| self.assertEqual(msg, MSG) |
| |
| def _testFullRead(self): |
| self.cli_file.write(MSG) |
| self.cli_file.close() |
| |
| def testUnbufferedRead(self): |
| # Performing unbuffered file read test |
| buf = '' |
| while 1: |
| char = self.serv_file.read(1) |
| if not char: |
| break |
| buf += char |
| self.assertEqual(buf, MSG) |
| |
| def _testUnbufferedRead(self): |
| self.cli_file.write(MSG) |
| self.cli_file.flush() |
| |
| def testReadline(self): |
| # Performing file readline test |
| line = self.serv_file.readline() |
| self.assertEqual(line, MSG) |
| |
| def _testReadline(self): |
| self.cli_file.write(MSG) |
| self.cli_file.flush() |
| |
| class UnbufferedFileObjectClassTestCase(FileObjectClassTestCase): |
| |
| """Repeat the tests from FileObjectClassTestCase with bufsize==0. |
| |
| In this case (and in this case only), it should be possible to |
| create a file object, read a line from it, create another file |
| object, read another line from it, without loss of data in the |
| first file object's buffer. Note that httplib relies on this |
| when reading multiple requests from the same socket.""" |
| |
| bufsize = 0 # Use unbuffered mode |
| |
| def testUnbufferedReadline(self): |
| # Read a line, create a new file object, read another line with it |
| line = self.serv_file.readline() # first line |
| self.assertEqual(line, "A. " + MSG) # first line |
| self.serv_file = self.cli_conn.makefile('rb', 0) |
| line = self.serv_file.readline() # second line |
| self.assertEqual(line, "B. " + MSG) # second line |
| |
| def _testUnbufferedReadline(self): |
| self.cli_file.write("A. " + MSG) |
| self.cli_file.write("B. " + MSG) |
| self.cli_file.flush() |
| |
| class LineBufferedFileObjectClassTestCase(FileObjectClassTestCase): |
| |
| bufsize = 1 # Default-buffered for reading; line-buffered for writing |
| |
| |
| class SmallBufferedFileObjectClassTestCase(FileObjectClassTestCase): |
| |
| bufsize = 2 # Exercise the buffering code |
| |
| class TCPTimeoutTest(SocketTCPTest): |
| |
| def testTCPTimeout(self): |
| def raise_timeout(*args, **kwargs): |
| self.serv.settimeout(1.0) |
| self.serv.accept() |
| self.failUnlessRaises(socket.timeout, raise_timeout, |
| "Error generating a timeout exception (TCP)") |
| |
| def testTimeoutZero(self): |
| ok = False |
| try: |
| self.serv.settimeout(0.0) |
| foo = self.serv.accept() |
| except socket.timeout: |
| self.fail("caught timeout instead of error (TCP)") |
| except socket.error: |
| ok = True |
| except: |
| self.fail("caught unexpected exception (TCP)") |
| if not ok: |
| self.fail("accept() returned success when we did not expect it") |
| |
| class UDPTimeoutTest(SocketTCPTest): |
| |
| def testUDPTimeout(self): |
| def raise_timeout(*args, **kwargs): |
| self.serv.settimeout(1.0) |
| self.serv.recv(1024) |
| self.failUnlessRaises(socket.timeout, raise_timeout, |
| "Error generating a timeout exception (UDP)") |
| |
| def testTimeoutZero(self): |
| ok = False |
| try: |
| self.serv.settimeout(0.0) |
| foo = self.serv.recv(1024) |
| except socket.timeout: |
| self.fail("caught timeout instead of error (UDP)") |
| except socket.error: |
| ok = True |
| except: |
| self.fail("caught unexpected exception (UDP)") |
| if not ok: |
| self.fail("recv() returned success when we did not expect it") |
| |
| class TestExceptions(unittest.TestCase): |
| |
| def testExceptionTree(self): |
| self.assert_(issubclass(socket.error, Exception)) |
| self.assert_(issubclass(socket.herror, socket.error)) |
| self.assert_(issubclass(socket.gaierror, socket.error)) |
| self.assert_(issubclass(socket.timeout, socket.error)) |
| |
| |
| def test_main(): |
| tests = [GeneralModuleTests, BasicTCPTest, TCPTimeoutTest, TestExceptions] |
| if sys.platform != 'mac': |
| tests.extend([ BasicUDPTest, UDPTimeoutTest ]) |
| |
| tests.extend([ |
| NonBlockingTCPTests, |
| FileObjectClassTestCase, |
| UnbufferedFileObjectClassTestCase, |
| LineBufferedFileObjectClassTestCase, |
| SmallBufferedFileObjectClassTestCase |
| ]) |
| test_support.run_unittest(*tests) |
| |
| if __name__ == "__main__": |
| test_main() |