mirror of
https://github.com/python/cpython.git
synced 2024-12-14 20:34:12 +08:00
775 lines
25 KiB
Python
775 lines
25 KiB
Python
#!/usr/bin/env python
|
|
|
|
import unittest
|
|
from test import test_support
|
|
|
|
import socket
|
|
import select
|
|
import time
|
|
import thread, threading
|
|
import Queue
|
|
import sys
|
|
from weakref import proxy
|
|
|
|
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 test_weakref(self):
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
p = proxy(s)
|
|
self.assertEqual(p.fileno(), s.fileno())
|
|
s.close()
|
|
s = None
|
|
try:
|
|
p.fileno()
|
|
except ReferenceError:
|
|
pass
|
|
else:
|
|
self.fail('Socket proxy still exists')
|
|
|
|
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
|
|
for proto in ("telnet", "ssh", "www", "ftp"):
|
|
try:
|
|
socket.getservbyname(proto, 'tcp')
|
|
break
|
|
except socket.error:
|
|
pass
|
|
try:
|
|
socket.getservbyname(proto, 'udp')
|
|
break
|
|
except socket.error:
|
|
pass
|
|
else:
|
|
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.assert_(self.serv_file.closed)
|
|
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.assert_(self.cli_file.closed)
|
|
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()
|
|
|
|
def testClosedAttr(self):
|
|
self.assert_(not self.serv_file.closed)
|
|
|
|
def _testClosedAttr(self):
|
|
self.assert_(not self.cli_file.closed)
|
|
|
|
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()
|