"""
    Most of this test module has been copied directly from the
    cpython 2.3 test_socket.py module. I have reduced it to only
    timeout related tests, and added a couple of new tests.
"""

from __future__ import nested_scopes

#
# This test suite should run on cpython 2.3+ and jython 2.1+.
#
# Ideally, it should generate identical results on both platforms.
# However, testTimeoutZero will give failures on cpython, because
# it is not possible to emulate the behaviour of settimeout(0) on 
# pre-java.nio JVMs. See notes below, beside the code for this test.
#

import os

try:
    True
except NameError:
    True, False = 1, 0

import socket

import unittest
import thread
import threading
import time
import Queue

PORT = 50007
HOST = 'localhost'
UNREACHABLE_HOST= '192.168.1.1'

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 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):
        try:
            self.serv.close()
        except Exception, x:
            pass 
        self.serv = None

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 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)

# Let the tests begin

class GeneralModuleTests(unittest.TestCase):

    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")

class ServerTimeoutTest(SocketTCPTest):

    def testAcceptTimeout(self):
        ok = False
        try:
            self.serv.settimeout(1.0)
            foo = self.serv.accept()
        except socket.timeout:
            ok = True
        except socket.error:
            self.fail("caught error instead of timeout (TCP)")
        except Exception, x:
            self.fail("caught unexpected exception (TCP): %s" % str(x))
        if not ok:
            self.fail("accept() returned success when we did not expect it")

    def testAcceptTimeoutDefaultOverride(self):
        ok = False
        try:
            self.serv.settimeout(1.0)
            # This should NOT override the timeout set on the socket
            # The test should just hang if it does
            socket.setdefaulttimeout(None)
            foo = self.serv.accept()
        except socket.timeout:
            ok = True
        except socket.error:
            self.fail("caught error instead of timeout (TCP)")
        except Exception, x:
            self.fail("caught unexpected exception (TCP): %s" % str(x))
        if not ok:
            self.fail("accept() returned success when we did not expect it")

    def testTimeoutZero(self):
        """
            This test will fail on cpython 2.3+. The semantic for settimeout(0)
            on cpython is "s.settimeout(0.0) is equivalent to s.setblocking(0)".
            But non-blocking socket I/O is not available on pre-1.4 JVMs, so the
            closest that can be achieved is to use the smallest possible timeout,
            i.e. 1 millisecond. This cpython test has been modified to reflect this
            difference.
        """
        ok = False
        try:
            self.serv.settimeout(0.0)
            foo = self.serv.accept()
        except socket.timeout:
            # self.fail("caught timeout instead of error (TCP)") # cpython behaviour
            ok = True
        except socket.error: 
            # ok = True # cpython behaviour
            self.fail("caught error instead of timeout (TCP)")
        except Exception, x:
            self.fail("caught unexpected exception (TCP): %s" % str(x))
        if not ok:
            self.fail("accept() returned success when we did not expect it")

class ClientTimeoutTest(SocketTCPTest):

    def testConnectTimeout(self):
        ok = False
        try:
            client = socket.socket()
            client.settimeout(1.0)
            client.connect((UNREACHABLE_HOST, PORT))
        except socket.timeout:
            ok = True
        except socket.error, se:
            self.fail("caught error instead of timeout (TCP): %s" % str(se))
        except Exception, x:
            self.fail("caught unexpected exception (TCP): %s" % str(x))
        if not ok:
            self.fail("connect() returned success when we did not expect it")

    def testConnectTimeoutDefaultTimeout(self):
        ok = False
        try:
            client = socket.socket()
            client.settimeout(1.0)
            # This should NOT override the timeout set on the socket
            socket.setdefaulttimeout(None)
            client.connect((UNREACHABLE_HOST, PORT))
        except socket.timeout:
            ok = True
        except socket.error, se:
            self.fail("caught error instead of timeout (TCP): %s" % str(se))
        except Exception, x:
            self.fail("caught unexpected exception (TCP): %s" % str(x))
        if not ok:
            self.fail("connect() returned success when we did not expect it")

class RecvTimeoutTest(SocketConnectedTest):

    def testTimeoutServerRead(self):
        ok = False
        try:
            self.cli_conn.settimeout(1.0)
            data = self.cli_conn.recv(1024)
        except socket.timeout:
            ok = True
        except socket.error:
            self.fail("caught error instead of timeout (TCP)")
        except Exception, x:
            self.fail("caught unexpected exception (TCP): %s" % str(x))
        if not ok:
            self.fail("server.recv() returned success when we did not expect it")

    def _testTimeoutServerRead(self):
        # Client side simply waits, so that the socket is closed prematurely.
        time.sleep(2.0)

    def testTimeoutClientRead(self):
        # Server side simply waits, so that the socket is closed prematurely.
        time.sleep(2.0)

    def _testTimeoutClientRead(self):
        ok = False
        try:
            self.serv_conn.settimeout(1.0)
            data = self.serv_conn.recv(1024)
        except socket.timeout:
            ok = True
        except socket.error:
            self.fail("caught error instead of timeout (TCP)")
        except Exception, x:
            self.fail("caught unexpected exception (TCP): %s" % str(x))
        if not ok:
            self.fail("client.recv() returned success when we did not expect it")

def test_main():
    tests = [
#    	GeneralModuleTests, 
#    	BasicTCPTest, 
    	TCPTimeoutTest, 
#   	TestExceptions,
    	]
    for t in tests:
    	print "Running test: %s" % t
    	test_support.run_unittest(t)

if __name__ == "__main__":
    print "Assuming that host '%s' is unreachable" % UNREACHABLE_HOST
    print "If host '%s' is reachable, you may get test failures" % UNREACHABLE_HOST
    unittest.main()
    # test_main()