d3c884d4ea
and age of rampant computer breakins I imagine there are plenty of systems with telnet disabled. Successful check of at least one getservbyname() call is required for success
572 lines
18 KiB
Python
572 lines
18 KiB
Python
#!/usr/bin/env python
|
|
|
|
import unittest
|
|
from test import test_support
|
|
|
|
import socket
|
|
import select
|
|
import time
|
|
import thread, threading
|
|
import Queue
|
|
|
|
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 that 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()
|
|
ip = socket.gethostbyname(hostname)
|
|
self.assert_(ip.find('.') >= 0, "Error resolving host to ip.")
|
|
hname, aliases, ipaddrs = socket.gethostbyaddr(ip)
|
|
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):
|
|
for func in socket.htonl, socket.ntohl:
|
|
for i in (0, 1, 0xffff0000, 2L):
|
|
self.assertEqual(i, func(func(i)))
|
|
|
|
biglong = 2**32L - 1
|
|
swapped = func(biglong)
|
|
self.assert_(swapped == biglong or swapped == -1)
|
|
self.assertRaises(OverflowError, func, 2L**34)
|
|
|
|
def testGetServByName(self):
|
|
"""Testing getservbyname()."""
|
|
if hasattr(socket, '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")
|
|
|
|
# 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."""
|
|
while 1:
|
|
read = self.cli_conn.recv(1024)
|
|
if not read:
|
|
break
|
|
self.assert_(len(read) == 1024, "Error performing sendall.")
|
|
read = filter(lambda x: x == 'f', read)
|
|
self.assert_(len(read) == 1024, "Error performing sendall.")
|
|
|
|
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):
|
|
|
|
def __init__(self, methodName='runTest'):
|
|
SocketConnectedTest.__init__(self, methodName=methodName)
|
|
|
|
def setUp(self):
|
|
SocketConnectedTest.setUp(self)
|
|
self.serv_file = socket._fileobject(self.cli_conn, 'rb', 8192)
|
|
|
|
def tearDown(self):
|
|
self.serv_file.close()
|
|
self.serv_file = None
|
|
SocketConnectedTest.tearDown(self)
|
|
|
|
def clientSetUp(self):
|
|
SocketConnectedTest.clientSetUp(self)
|
|
self.cli_file = socket._fileobject(self.serv_conn, 'rb', 8192)
|
|
|
|
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 testUnbufferedRead(self):
|
|
"""Performing unbuffered file read test."""
|
|
buf = ''
|
|
while 1:
|
|
char = self.serv_file.read(1)
|
|
self.failIf(not char)
|
|
buf += char
|
|
if buf == MSG:
|
|
break
|
|
|
|
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 test_main():
|
|
suite = unittest.TestSuite()
|
|
suite.addTest(unittest.makeSuite(GeneralModuleTests))
|
|
suite.addTest(unittest.makeSuite(BasicTCPTest))
|
|
suite.addTest(unittest.makeSuite(BasicUDPTest))
|
|
suite.addTest(unittest.makeSuite(NonBlockingTCPTests))
|
|
suite.addTest(unittest.makeSuite(FileObjectClassTestCase))
|
|
test_support.run_suite(suite)
|
|
|
|
if __name__ == "__main__":
|
|
test_main()
|