0001 #!/usr/bin/env python 0002 0003 import unittest 0004 from test import test_support 0005 0006 import socket 0007 import select 0008 import time 0009 import thread, threading 0010 import Queue 0011 import sys 0012 from weakref import proxy 0013 0014 PORT = 50007 0015 HOST = 'localhost' 0016 MSG = 'Michael Gilfix was here\n' 0017 0018 class SocketTCPTest(unittest.TestCase): 0019 0020 def setUp(self): 0021 self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 0022 self.serv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) 0023 self.serv.bind((HOST, PORT)) 0024 self.serv.listen(1) 0025 0026 def tearDown(self): 0027 self.serv.close() 0028 self.serv = None 0029 0030 class SocketUDPTest(unittest.TestCase): 0031 0032 def setUp(self): 0033 self.serv = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) 0034 self.serv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) 0035 self.serv.bind((HOST, PORT)) 0036 0037 def tearDown(self): 0038 self.serv.close() 0039 self.serv = None 0040 0041 class ThreadableTest: 0042 """Threadable Test class 0043 0044 The ThreadableTest class makes it easy to create a threaded 0045 client/server pair from an existing unit test. To create a 0046 new threaded class from an existing unit test, use multiple 0047 inheritance: 0048 0049 class NewClass (OldClass, ThreadableTest): 0050 pass 0051 0052 This class defines two new fixture functions with obvious 0053 purposes for overriding: 0054 0055 clientSetUp () 0056 clientTearDown () 0057 0058 Any new test functions within the class must then define 0059 tests in pairs, where the test name is preceeded with a 0060 '_' to indicate the client portion of the test. Ex: 0061 0062 def testFoo(self): 0063 # Server portion 0064 0065 def _testFoo(self): 0066 # Client portion 0067 0068 Any exceptions raised by the clients during their tests 0069 are caught and transferred to the main thread to alert 0070 the testing framework. 0071 0072 Note, the server setup function cannot call any blocking 0073 functions that rely on the client thread during setup, 0074 unless serverExplicityReady() is called just before 0075 the blocking call (such as in setting up a client/server 0076 connection and performing the accept() in setUp(). 0077 """ 0078 0079 def __init__(self): 0080 # Swap the true setup function 0081 self.__setUp = self.setUp 0082 self.__tearDown = self.tearDown 0083 self.setUp = self._setUp 0084 self.tearDown = self._tearDown 0085 0086 def serverExplicitReady(self): 0087 """This method allows the server to explicitly indicate that 0088 it wants the client thread to proceed. This is useful if the 0089 server is about to execute a blocking routine that is 0090 dependent upon the client thread during its setup routine.""" 0091 self.server_ready.set() 0092 0093 def _setUp(self): 0094 self.server_ready = threading.Event() 0095 self.client_ready = threading.Event() 0096 self.done = threading.Event() 0097 self.queue = Queue.Queue(1) 0098 0099 # Do some munging to start the client test. 0100 methodname = self.id() 0101 i = methodname.rfind('.') 0102 methodname = methodname[i+1:] 0103 test_method = getattr(self, '_' + methodname) 0104 self.client_thread = thread.start_new_thread( 0105 self.clientRun, (test_method,)) 0106 0107 self.__setUp() 0108 if not self.server_ready.isSet(): 0109 self.server_ready.set() 0110 self.client_ready.wait() 0111 0112 def _tearDown(self): 0113 self.__tearDown() 0114 self.done.wait() 0115 0116 if not self.queue.empty(): 0117 msg = self.queue.get() 0118 self.fail(msg) 0119 0120 def clientRun(self, test_func): 0121 self.server_ready.wait() 0122 self.client_ready.set() 0123 self.clientSetUp() 0124 if not callable(test_func): 0125 raise TypeError, "test_func must be a callable function" 0126 try: 0127 test_func() 0128 except Exception, strerror: 0129 self.queue.put(strerror) 0130 self.clientTearDown() 0131 0132 def clientSetUp(self): 0133 raise NotImplementedError, "clientSetUp must be implemented." 0134 0135 def clientTearDown(self): 0136 self.done.set() 0137 thread.exit() 0138 0139 class ThreadedTCPSocketTest(SocketTCPTest, ThreadableTest): 0140 0141 def __init__(self, methodName='runTest'): 0142 SocketTCPTest.__init__(self, methodName=methodName) 0143 ThreadableTest.__init__(self) 0144 0145 def clientSetUp(self): 0146 self.cli = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 0147 0148 def clientTearDown(self): 0149 self.cli.close() 0150 self.cli = None 0151 ThreadableTest.clientTearDown(self) 0152 0153 class ThreadedUDPSocketTest(SocketUDPTest, ThreadableTest): 0154 0155 def __init__(self, methodName='runTest'): 0156 SocketUDPTest.__init__(self, methodName=methodName) 0157 ThreadableTest.__init__(self) 0158 0159 def clientSetUp(self): 0160 self.cli = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) 0161 0162 class SocketConnectedTest(ThreadedTCPSocketTest): 0163 0164 def __init__(self, methodName='runTest'): 0165 ThreadedTCPSocketTest.__init__(self, methodName=methodName) 0166 0167 def setUp(self): 0168 ThreadedTCPSocketTest.setUp(self) 0169 # Indicate explicitly we're ready for the client thread to 0170 # proceed and then perform the blocking call to accept 0171 self.serverExplicitReady() 0172 conn, addr = self.serv.accept() 0173 self.cli_conn = conn 0174 0175 def tearDown(self): 0176 self.cli_conn.close() 0177 self.cli_conn = None 0178 ThreadedTCPSocketTest.tearDown(self) 0179 0180 def clientSetUp(self): 0181 ThreadedTCPSocketTest.clientSetUp(self) 0182 self.cli.connect((HOST, PORT)) 0183 self.serv_conn = self.cli 0184 0185 def clientTearDown(self): 0186 self.serv_conn.close() 0187 self.serv_conn = None 0188 ThreadedTCPSocketTest.clientTearDown(self) 0189 0190 class SocketPairTest(unittest.TestCase, ThreadableTest): 0191 0192 def __init__(self, methodName='runTest'): 0193 unittest.TestCase.__init__(self, methodName=methodName) 0194 ThreadableTest.__init__(self) 0195 0196 def setUp(self): 0197 self.serv, self.cli = socket.socketpair() 0198 0199 def tearDown(self): 0200 self.serv.close() 0201 self.serv = None 0202 0203 def clientSetUp(self): 0204 pass 0205 0206 def clientTearDown(self): 0207 self.cli.close() 0208 self.cli = None 0209 ThreadableTest.clientTearDown(self) 0210 0211 0212 ####################################################################### 0213 ## Begin Tests 0214 0215 class GeneralModuleTests(unittest.TestCase): 0216 0217 def test_weakref(self): 0218 s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 0219 p = proxy(s) 0220 self.assertEqual(p.fileno(), s.fileno()) 0221 s.close() 0222 s = None 0223 try: 0224 p.fileno() 0225 except ReferenceError: 0226 pass 0227 else: 0228 self.fail('Socket proxy still exists') 0229 0230 def testSocketError(self): 0231 # Testing socket module exceptions 0232 def raise_error(*args, **kwargs): 0233 raise socket.error 0234 def raise_herror(*args, **kwargs): 0235 raise socket.herror 0236 def raise_gaierror(*args, **kwargs): 0237 raise socket.gaierror 0238 self.failUnlessRaises(socket.error, raise_error, 0239 "Error raising socket exception.") 0240 self.failUnlessRaises(socket.error, raise_herror, 0241 "Error raising socket exception.") 0242 self.failUnlessRaises(socket.error, raise_gaierror, 0243 "Error raising socket exception.") 0244 0245 def testCrucialConstants(self): 0246 # Testing for mission critical constants 0247 socket.AF_INET 0248 socket.SOCK_STREAM 0249 socket.SOCK_DGRAM 0250 socket.SOCK_RAW 0251 socket.SOCK_RDM 0252 socket.SOCK_SEQPACKET 0253 socket.SOL_SOCKET 0254 socket.SO_REUSEADDR 0255 0256 def testHostnameRes(self): 0257 # Testing hostname resolution mechanisms 0258 hostname = socket.gethostname() 0259 try: 0260 ip = socket.gethostbyname(hostname) 0261 except socket.error: 0262 # Probably name lookup wasn't set up right; skip this test 0263 return 0264 self.assert_(ip.find('.') >= 0, "Error resolving host to ip.") 0265 try: 0266 hname, aliases, ipaddrs = socket.gethostbyaddr(ip) 0267 except socket.error: 0268 # Probably a similar problem as above; skip this test 0269 return 0270 all_host_names = [hname] + aliases 0271 fqhn = socket.getfqdn() 0272 if not fqhn in all_host_names: 0273 self.fail("Error testing host resolution mechanisms.") 0274 0275 def testRefCountGetNameInfo(self): 0276 # Testing reference count for getnameinfo 0277 import sys 0278 if hasattr(sys, "getrefcount"): 0279 try: 0280 # On some versions, this loses a reference 0281 orig = sys.getrefcount(__name__) 0282 socket.getnameinfo(__name__,0) 0283 except SystemError: 0284 if sys.getrefcount(__name__) <> orig: 0285 self.fail("socket.getnameinfo loses a reference") 0286 0287 def testInterpreterCrash(self): 0288 # Making sure getnameinfo doesn't crash the interpreter 0289 try: 0290 # On some versions, this crashes the interpreter. 0291 socket.getnameinfo(('x', 0, 0, 0), 0) 0292 except socket.error: 0293 pass 0294 0295 def testNtoH(self): 0296 # This just checks that htons etc. are their own inverse, 0297 # when looking at the lower 16 or 32 bits. 0298 sizes = {socket.htonl: 32, socket.ntohl: 32, 0299 socket.htons: 16, socket.ntohs: 16} 0300 for func, size in sizes.items(): 0301 mask = (1L<<size) - 1 0302 for i in (0, 1, 0xffff, ~0xffff, 2, 0x01234567, 0x76543210): 0303 self.assertEqual(i & mask, func(func(i&mask)) & mask) 0304 0305 swapped = func(mask) 0306 self.assertEqual(swapped & mask, mask) 0307 self.assertRaises(OverflowError, func, 1L<<34) 0308 0309 def testGetServBy(self): 0310 eq = self.assertEqual 0311 # Find one service that exists, then check all the related interfaces. 0312 # I've ordered this by protocols that have both a tcp and udp 0313 # protocol, at least for modern Linuxes. 0314 if sys.platform in ('freebsd4', 'freebsd5', 'freebsd6', 'darwin'): 0315 # avoid the 'echo' service on this platform, as there is an 0316 # assumption breaking non-standard port/protocol entry 0317 services = ('daytime', 'qotd', 'domain') 0318 else: 0319 services = ('echo', 'daytime', 'domain') 0320 for service in services: 0321 try: 0322 port = socket.getservbyname(service, 'tcp') 0323 break 0324 except socket.error: 0325 pass 0326 else: 0327 raise socket.error 0328 # Try same call with optional protocol omitted 0329 port2 = socket.getservbyname(service) 0330 eq(port, port2) 0331 # Try udp, but don't barf it it doesn't exist 0332 try: 0333 udpport = socket.getservbyname(service, 'udp') 0334 except socket.error: 0335 udpport = None 0336 else: 0337 eq(udpport, port) 0338 # Now make sure the lookup by port returns the same service name 0339 eq(socket.getservbyport(port2), service) 0340 eq(socket.getservbyport(port, 'tcp'), service) 0341 if udpport is not None: 0342 eq(socket.getservbyport(udpport, 'udp'), service) 0343 0344 def testDefaultTimeout(self): 0345 # Testing default timeout 0346 # The default timeout should initially be None 0347 self.assertEqual(socket.getdefaulttimeout(), None) 0348 s = socket.socket() 0349 self.assertEqual(s.gettimeout(), None) 0350 s.close() 0351 0352 # Set the default timeout to 10, and see if it propagates 0353 socket.setdefaulttimeout(10) 0354 self.assertEqual(socket.getdefaulttimeout(), 10) 0355 s = socket.socket() 0356 self.assertEqual(s.gettimeout(), 10) 0357 s.close() 0358 0359 # Reset the default timeout to None, and see if it propagates 0360 socket.setdefaulttimeout(None) 0361 self.assertEqual(socket.getdefaulttimeout(), None) 0362 s = socket.socket() 0363 self.assertEqual(s.gettimeout(), None) 0364 s.close() 0365 0366 # Check that setting it to an invalid value raises ValueError 0367 self.assertRaises(ValueError, socket.setdefaulttimeout, -1) 0368 0369 # Check that setting it to an invalid type raises TypeError 0370 self.assertRaises(TypeError, socket.setdefaulttimeout, "spam") 0371 0372 def testIPv4toString(self): 0373 if not hasattr(socket, 'inet_pton'): 0374 return # No inet_pton() on this platform 0375 from socket import inet_aton as f, inet_pton, AF_INET 0376 g = lambda a: inet_pton(AF_INET, a) 0377 0378 self.assertEquals('\x00\x00\x00\x00', f('0.0.0.0')) 0379 self.assertEquals('\xff\x00\xff\x00', f('255.0.255.0')) 0380 self.assertEquals('\xaa\xaa\xaa\xaa', f('170.170.170.170')) 0381 self.assertEquals('\x01\x02\x03\x04', f('1.2.3.4')) 0382 0383 self.assertEquals('\x00\x00\x00\x00', g('0.0.0.0')) 0384 self.assertEquals('\xff\x00\xff\x00', g('255.0.255.0')) 0385 self.assertEquals('\xaa\xaa\xaa\xaa', g('170.170.170.170')) 0386 0387 def testIPv6toString(self): 0388 if not hasattr(socket, 'inet_pton'): 0389 return # No inet_pton() on this platform 0390 try: 0391 from socket import inet_pton, AF_INET6, has_ipv6 0392 if not has_ipv6: 0393 return 0394 except ImportError: 0395 return 0396 f = lambda a: inet_pton(AF_INET6, a) 0397 0398 self.assertEquals('\x00' * 16, f('::')) 0399 self.assertEquals('\x00' * 16, f('0::0')) 0400 self.assertEquals('\x00\x01' + '\x00' * 14, f('1::')) 0401 self.assertEquals( 0402 '\x45\xef\x76\xcb\x00\x1a\x56\xef\xaf\xeb\x0b\xac\x19\x24\xae\xae', 0403 f('45ef:76cb:1a:56ef:afeb:bac:1924:aeae') 0404 ) 0405 0406 def testStringToIPv4(self): 0407 if not hasattr(socket, 'inet_ntop'): 0408 return # No inet_ntop() on this platform 0409 from socket import inet_ntoa as f, inet_ntop, AF_INET 0410 g = lambda a: inet_ntop(AF_INET, a) 0411 0412 self.assertEquals('1.0.1.0', f('\x01\x00\x01\x00')) 0413 self.assertEquals('170.85.170.85', f('\xaa\x55\xaa\x55')) 0414 self.assertEquals('255.255.255.255', f('\xff\xff\xff\xff')) 0415 self.assertEquals('1.2.3.4', f('\x01\x02\x03\x04')) 0416 0417 self.assertEquals('1.0.1.0', g('\x01\x00\x01\x00')) 0418 self.assertEquals('170.85.170.85', g('\xaa\x55\xaa\x55')) 0419 self.assertEquals('255.255.255.255', g('\xff\xff\xff\xff')) 0420 0421 def testStringToIPv6(self): 0422 if not hasattr(socket, 'inet_ntop'): 0423 return # No inet_ntop() on this platform 0424 try: 0425 from socket import inet_ntop, AF_INET6, has_ipv6 0426 if not has_ipv6: 0427 return 0428 except ImportError: 0429 return 0430 f = lambda a: inet_ntop(AF_INET6, a) 0431 0432 self.assertEquals('::', f('\x00' * 16)) 0433 self.assertEquals('::1', f('\x00' * 15 + '\x01')) 0434 self.assertEquals( 0435 'aef:b01:506:1001:ffff:9997:55:170', 0436 f('\x0a\xef\x0b\x01\x05\x06\x10\x01\xff\xff\x99\x97\x00\x55\x01\x70') 0437 ) 0438 0439 # XXX The following don't test module-level functionality... 0440 0441 def testSockName(self): 0442 # Testing getsockname() 0443 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 0444 sock.bind(("0.0.0.0", PORT+1)) 0445 name = sock.getsockname() 0446 self.assertEqual(name, ("0.0.0.0", PORT+1)) 0447 0448 def testGetSockOpt(self): 0449 # Testing getsockopt() 0450 # We know a socket should start without reuse==0 0451 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 0452 reuse = sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR) 0453 self.failIf(reuse != 0, "initial mode is reuse") 0454 0455 def testSetSockOpt(self): 0456 # Testing setsockopt() 0457 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 0458 sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) 0459 reuse = sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR) 0460 self.failIf(reuse == 0, "failed to set reuse mode") 0461 0462 def testSendAfterClose(self): 0463 # testing send() after close() with timeout 0464 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 0465 sock.settimeout(1) 0466 sock.close() 0467 self.assertRaises(socket.error, sock.send, "spam") 0468 0469 class BasicTCPTest(SocketConnectedTest): 0470 0471 def __init__(self, methodName='runTest'): 0472 SocketConnectedTest.__init__(self, methodName=methodName) 0473 0474 def testRecv(self): 0475 # Testing large receive over TCP 0476 msg = self.cli_conn.recv(1024) 0477 self.assertEqual(msg, MSG) 0478 0479 def _testRecv(self): 0480 self.serv_conn.send(MSG) 0481 0482 def testOverFlowRecv(self): 0483 # Testing receive in chunks over TCP 0484 seg1 = self.cli_conn.recv(len(MSG) - 3) 0485 seg2 = self.cli_conn.recv(1024) 0486 msg = seg1 + seg2 0487 self.assertEqual(msg, MSG) 0488 0489 def _testOverFlowRecv(self): 0490 self.serv_conn.send(MSG) 0491 0492 def testRecvFrom(self): 0493 # Testing large recvfrom() over TCP 0494 msg, addr = self.cli_conn.recvfrom(1024) 0495 self.assertEqual(msg, MSG) 0496 0497 def _testRecvFrom(self): 0498 self.serv_conn.send(MSG) 0499 0500 def testOverFlowRecvFrom(self): 0501 # Testing recvfrom() in chunks over TCP 0502 seg1, addr = self.cli_conn.recvfrom(len(MSG)-3) 0503 seg2, addr = self.cli_conn.recvfrom(1024) 0504 msg = seg1 + seg2 0505 self.assertEqual(msg, MSG) 0506 0507 def _testOverFlowRecvFrom(self): 0508 self.serv_conn.send(MSG) 0509 0510 def testSendAll(self): 0511 # Testing sendall() with a 2048 byte string over TCP 0512 msg = '' 0513 while 1: 0514 read = self.cli_conn.recv(1024) 0515 if not read: 0516 break 0517 msg += read 0518 self.assertEqual(msg, 'f' * 2048) 0519 0520 def _testSendAll(self): 0521 big_chunk = 'f' * 2048 0522 self.serv_conn.sendall(big_chunk) 0523 0524 def testFromFd(self): 0525 # Testing fromfd() 0526 if not hasattr(socket, "fromfd"): 0527 return # On Windows, this doesn't exist 0528 fd = self.cli_conn.fileno() 0529 sock = socket.fromfd(fd, socket.AF_INET, socket.SOCK_STREAM) 0530 msg = sock.recv(1024) 0531 self.assertEqual(msg, MSG) 0532 0533 def _testFromFd(self): 0534 self.serv_conn.send(MSG) 0535 0536 def testShutdown(self): 0537 # Testing shutdown() 0538 msg = self.cli_conn.recv(1024) 0539 self.assertEqual(msg, MSG) 0540 0541 def _testShutdown(self): 0542 self.serv_conn.send(MSG) 0543 self.serv_conn.shutdown(2) 0544 0545 class BasicUDPTest(ThreadedUDPSocketTest): 0546 0547 def __init__(self, methodName='runTest'): 0548 ThreadedUDPSocketTest.__init__(self, methodName=methodName) 0549 0550 def testSendtoAndRecv(self): 0551 # Testing sendto() and Recv() over UDP 0552 msg = self.serv.recv(len(MSG)) 0553 self.assertEqual(msg, MSG) 0554 0555 def _testSendtoAndRecv(self): 0556 self.cli.sendto(MSG, 0, (HOST, PORT)) 0557 0558 def testRecvFrom(self): 0559 # Testing recvfrom() over UDP 0560 msg, addr = self.serv.recvfrom(len(MSG)) 0561 self.assertEqual(msg, MSG) 0562 0563 def _testRecvFrom(self): 0564 self.cli.sendto(MSG, 0, (HOST, PORT)) 0565 0566 class BasicSocketPairTest(SocketPairTest): 0567 0568 def __init__(self, methodName='runTest'): 0569 SocketPairTest.__init__(self, methodName=methodName) 0570 0571 def testRecv(self): 0572 msg = self.serv.recv(1024) 0573 self.assertEqual(msg, MSG) 0574 0575 def _testRecv(self): 0576 self.cli.send(MSG) 0577 0578 def testSend(self): 0579 self.serv.send(MSG) 0580 0581 def _testSend(self): 0582 msg = self.cli.recv(1024) 0583 self.assertEqual(msg, MSG) 0584 0585 class NonBlockingTCPTests(ThreadedTCPSocketTest): 0586 0587 def __init__(self, methodName='runTest'): 0588 ThreadedTCPSocketTest.__init__(self, methodName=methodName) 0589 0590 def testSetBlocking(self): 0591 # Testing whether set blocking works 0592 self.serv.setblocking(0) 0593 start = time.time() 0594 try: 0595 self.serv.accept() 0596 except socket.error: 0597 pass 0598 end = time.time() 0599 self.assert_((end - start) < 1.0, "Error setting non-blocking mode.") 0600 0601 def _testSetBlocking(self): 0602 pass 0603 0604 def testAccept(self): 0605 # Testing non-blocking accept 0606 self.serv.setblocking(0) 0607 try: 0608 conn, addr = self.serv.accept() 0609 except socket.error: 0610 pass 0611 else: 0612 self.fail("Error trying to do non-blocking accept.") 0613 read, write, err = select.select([self.serv], [], []) 0614 if self.serv in read: 0615 conn, addr = self.serv.accept() 0616 else: 0617 self.fail("Error trying to do accept after select.") 0618 0619 def _testAccept(self): 0620 time.sleep(0.1) 0621 self.cli.connect((HOST, PORT)) 0622 0623 def testConnect(self): 0624 # Testing non-blocking connect 0625 conn, addr = self.serv.accept() 0626 0627 def _testConnect(self): 0628 self.cli.settimeout(10) 0629 self.cli.connect((HOST, PORT)) 0630 0631 def testRecv(self): 0632 # Testing non-blocking recv 0633 conn, addr = self.serv.accept() 0634 conn.setblocking(0) 0635 try: 0636 msg = conn.recv(len(MSG)) 0637 except socket.error: 0638 pass 0639 else: 0640 self.fail("Error trying to do non-blocking recv.") 0641 read, write, err = select.select([conn], [], []) 0642 if conn in read: 0643 msg = conn.recv(len(MSG)) 0644 self.assertEqual(msg, MSG) 0645 else: 0646 self.fail("Error during select call to non-blocking socket.") 0647 0648 def _testRecv(self): 0649 self.cli.connect((HOST, PORT)) 0650 time.sleep(0.1) 0651 self.cli.send(MSG) 0652 0653 class FileObjectClassTestCase(SocketConnectedTest): 0654 0655 bufsize = -1 # Use default buffer size 0656 0657 def __init__(self, methodName='runTest'): 0658 SocketConnectedTest.__init__(self, methodName=methodName) 0659 0660 def setUp(self): 0661 SocketConnectedTest.setUp(self) 0662 self.serv_file = self.cli_conn.makefile('rb', self.bufsize) 0663 0664 def tearDown(self): 0665 self.serv_file.close() 0666 self.assert_(self.serv_file.closed) 0667 self.serv_file = None 0668 SocketConnectedTest.tearDown(self) 0669 0670 def clientSetUp(self): 0671 SocketConnectedTest.clientSetUp(self) 0672 self.cli_file = self.serv_conn.makefile('wb') 0673 0674 def clientTearDown(self): 0675 self.cli_file.close() 0676 self.assert_(self.cli_file.closed) 0677 self.cli_file = None 0678 SocketConnectedTest.clientTearDown(self) 0679 0680 def testSmallRead(self): 0681 # Performing small file read test 0682 first_seg = self.serv_file.read(len(MSG)-3) 0683 second_seg = self.serv_file.read(3) 0684 msg = first_seg + second_seg 0685 self.assertEqual(msg, MSG) 0686 0687 def _testSmallRead(self): 0688 self.cli_file.write(MSG) 0689 self.cli_file.flush() 0690 0691 def testFullRead(self): 0692 # read until EOF 0693 msg = self.serv_file.read() 0694 self.assertEqual(msg, MSG) 0695 0696 def _testFullRead(self): 0697 self.cli_file.write(MSG) 0698 self.cli_file.close() 0699 0700 def testUnbufferedRead(self): 0701 # Performing unbuffered file read test 0702 buf = '' 0703 while 1: 0704 char = self.serv_file.read(1) 0705 if not char: 0706 break 0707 buf += char 0708 self.assertEqual(buf, MSG) 0709 0710 def _testUnbufferedRead(self): 0711 self.cli_file.write(MSG) 0712 self.cli_file.flush() 0713 0714 def testReadline(self): 0715 # Performing file readline test 0716 line = self.serv_file.readline() 0717 self.assertEqual(line, MSG) 0718 0719 def _testReadline(self): 0720 self.cli_file.write(MSG) 0721 self.cli_file.flush() 0722 0723 def testClosedAttr(self): 0724 self.assert_(not self.serv_file.closed) 0725 0726 def _testClosedAttr(self): 0727 self.assert_(not self.cli_file.closed) 0728 0729 class UnbufferedFileObjectClassTestCase(FileObjectClassTestCase): 0730 0731 """Repeat the tests from FileObjectClassTestCase with bufsize==0. 0732 0733 In this case (and in this case only), it should be possible to 0734 create a file object, read a line from it, create another file 0735 object, read another line from it, without loss of data in the 0736 first file object's buffer. Note that httplib relies on this 0737 when reading multiple requests from the same socket.""" 0738 0739 bufsize = 0 # Use unbuffered mode 0740 0741 def testUnbufferedReadline(self): 0742 # Read a line, create a new file object, read another line with it 0743 line = self.serv_file.readline() # first line 0744 self.assertEqual(line, "A. " + MSG) # first line 0745 self.serv_file = self.cli_conn.makefile('rb', 0) 0746 line = self.serv_file.readline() # second line 0747 self.assertEqual(line, "B. " + MSG) # second line 0748 0749 def _testUnbufferedReadline(self): 0750 self.cli_file.write("A. " + MSG) 0751 self.cli_file.write("B. " + MSG) 0752 self.cli_file.flush() 0753 0754 class LineBufferedFileObjectClassTestCase(FileObjectClassTestCase): 0755 0756 bufsize = 1 # Default-buffered for reading; line-buffered for writing 0757 0758 0759 class SmallBufferedFileObjectClassTestCase(FileObjectClassTestCase): 0760 0761 bufsize = 2 # Exercise the buffering code 0762 0763 class TCPTimeoutTest(SocketTCPTest): 0764 0765 def testTCPTimeout(self): 0766 def raise_timeout(*args, **kwargs): 0767 self.serv.settimeout(1.0) 0768 self.serv.accept() 0769 self.failUnlessRaises(socket.timeout, raise_timeout, 0770 "Error generating a timeout exception (TCP)") 0771 0772 def testTimeoutZero(self): 0773 ok = False 0774 try: 0775 self.serv.settimeout(0.0) 0776 foo = self.serv.accept() 0777 except socket.timeout: 0778 self.fail("caught timeout instead of error (TCP)") 0779 except socket.error: 0780 ok = True 0781 except: 0782 self.fail("caught unexpected exception (TCP)") 0783 if not ok: 0784 self.fail("accept() returned success when we did not expect it") 0785 0786 class UDPTimeoutTest(SocketTCPTest): 0787 0788 def testUDPTimeout(self): 0789 def raise_timeout(*args, **kwargs): 0790 self.serv.settimeout(1.0) 0791 self.serv.recv(1024) 0792 self.failUnlessRaises(socket.timeout, raise_timeout, 0793 "Error generating a timeout exception (UDP)") 0794 0795 def testTimeoutZero(self): 0796 ok = False 0797 try: 0798 self.serv.settimeout(0.0) 0799 foo = self.serv.recv(1024) 0800 except socket.timeout: 0801 self.fail("caught timeout instead of error (UDP)") 0802 except socket.error: 0803 ok = True 0804 except: 0805 self.fail("caught unexpected exception (UDP)") 0806 if not ok: 0807 self.fail("recv() returned success when we did not expect it") 0808 0809 class TestExceptions(unittest.TestCase): 0810 0811 def testExceptionTree(self): 0812 self.assert_(issubclass(socket.error, Exception)) 0813 self.assert_(issubclass(socket.herror, socket.error)) 0814 self.assert_(issubclass(socket.gaierror, socket.error)) 0815 self.assert_(issubclass(socket.timeout, socket.error)) 0816 0817 0818 def test_main(): 0819 tests = [GeneralModuleTests, BasicTCPTest, TCPTimeoutTest, TestExceptions] 0820 if sys.platform != 'mac': 0821 tests.extend([ BasicUDPTest, UDPTimeoutTest ]) 0822 0823 tests.extend([ 0824 NonBlockingTCPTests, 0825 FileObjectClassTestCase, 0826 UnbufferedFileObjectClassTestCase, 0827 LineBufferedFileObjectClassTestCase, 0828 SmallBufferedFileObjectClassTestCase 0829 ]) 0830 if hasattr(socket, "socketpair"): 0831 tests.append(BasicSocketPairTest) 0832 test_support.run_unittest(*tests) 0833 0834 if __name__ == "__main__": 0835 test_main() 0836
Generated by PyXR 0.9.4