~bzr-pqm/bzr/bzr.dev

# Copyright (C) 2010 Canonical Ltd

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA

import errno

import socket

import SocketServer

import threading

from bzrlib import (

    osutils,

    tests,

    )

from bzrlib.tests import test_server

def load_tests(basic_tests, module, loader):

    suite = loader.suiteClass()

    server_tests, remaining_tests = tests.split_suite_by_condition(

        basic_tests, tests.condition_isinstance(TestTCPServerInAThread))

    server_scenarios = [ (name, {'server_class': getattr(test_server, name)})

                         for name in

                         ('TestingTCPServer', 'TestingThreadingTCPServer')]

    tests.multiply_tests(server_tests, server_scenarios, suite)

    suite.addTest(remaining_tests)

    return suite

class TestThreadWithException(tests.TestCase):

    def test_start_and_join_smoke_test(self):

        def do_nothing():

            pass

        tt = test_server.ThreadWithException(target=do_nothing)

        tt.start()

        tt.join()

    def test_exception_is_re_raised(self):

        class MyException(Exception):

            pass

        def raise_my_exception():

            raise MyException()

        tt = test_server.ThreadWithException(target=raise_my_exception)

        tt.start()

        self.assertRaises(MyException, tt.join)

    def test_join_when_no_exception(self):

        resume = threading.Event()

        class MyException(Exception):

            pass

        def raise_my_exception():

            # Wait for the test to tell us to resume

            resume.wait()

            # Now we can raise

            raise MyException()

        tt = test_server.ThreadWithException(target=raise_my_exception)

        tt.start()

        tt.join(timeout=0)

        self.assertIs(None, tt.exception)

        resume.set()

        self.assertRaises(MyException, tt.join)

class TCPClient(object):

    def __init__(self):

        self.sock = None

    def connect(self, addr):

        if self.sock is not None:

            raise AssertionError('Already connected to %r'

                                 % (self.sock.getsockname(),))

        self.sock = osutils.connect_socket(addr)

    def disconnect(self):

        if self.sock is not None:

            try:

                self.sock.shutdown(socket.SHUT_RDWR)

                self.sock.close()

            except socket.error, e:

                if e[0] in (errno.EBADF, errno.ENOTCONN):

                    # Right, the socket is already down

                    pass

                else:

                    raise

            self.sock = None

    def write(self, s):

        return self.sock.sendall(s)

    def read(self, bufsize=4096):

        return self.sock.recv(bufsize)

class TCPConnectionHandler(SocketServer.StreamRequestHandler):

    def handle(self):

        self.done = False

        self.handle_connection()

        while not self.done:

            self.handle_connection()

    def handle_connection(self):

        req = self.rfile.readline()

        if not req:

            self.done = True

        elif req == 'ping\n':

            self.wfile.write('pong\n')

        else:

            raise ValueError('[%s] not understood' % req)

class TestTCPServerInAThread(tests.TestCase):

    # Set by load_tests()

    server_class = None

    def get_server(self, server_class=None, connection_handler_class=None):

        if server_class is not None:

            self.server_class = server_class

        if connection_handler_class is None:

            connection_handler_class = TCPConnectionHandler

        server =  test_server.TestingTCPServerInAThread(

            ('localhost', 0), self.server_class, connection_handler_class)

        server.start_server()

        self.addCleanup(server.stop_server)

        return server

    def get_client(self):

        client = TCPClient()

        self.addCleanup(client.disconnect)

        return client

    def get_server_connection(self, server, conn_rank):

        return server.server.clients[conn_rank]

    def assertClientAddr(self, client, server, conn_rank):

        conn = self.get_server_connection(server, conn_rank)

        self.assertEquals(client.sock.getsockname(), conn[1])

    def test_start_stop(self):

        server = self.get_server()

        client = self.get_client()

        server.stop_server()

        # since the server doesn't accept connections anymore attempting to

        # connect should fail

        client = self.get_client()

        self.assertRaises(socket.error,

                          client.connect, (server.host, server.port))

    def test_client_talks_server_respond(self):

        server = self.get_server()

        client = self.get_client()

        client.connect((server.host, server.port))

        self.assertIs(None, client.write('ping\n'))

        resp = client.read()

        self.assertClientAddr(client, server, 0)

        self.assertEquals('pong\n', resp)

    def test_server_fails_to_start(self):

        class CantStart(Exception):

            pass

        class CantStartServer(test_server.TestingTCPServer):

            def server_bind(self):

                raise CantStart()

        # The exception is raised in the main thread

        self.assertRaises(CantStart,

                          self.get_server, server_class=CantStartServer)

    def test_server_fails_while_serving_or_stopping(self):

        class CantConnect(Exception):

            pass

        class FailingConnectionHandler(TCPConnectionHandler):

            def handle(self):

                raise CantConnect()

        server = self.get_server(

            connection_handler_class=FailingConnectionHandler)

        # The server won't fail until a client connect

        client = self.get_client()

        client.connect((server.host, server.port))

        try:

            # Now we must force the server to answer by sending the request and

            # waiting for some answer. But since we don't control when the

            # server thread will be given cycles, we don't control either

            # whether our reads or writes may hang.

            client.sock.settimeout(0.1)

            client.write('ping\n')

            client.read()

        except socket.error:

            pass

        # Now the server has raised the exception in its own thread

        self.assertRaises(CantConnect, server.stop_server)

    def test_server_crash_while_responding(self):

        sync = threading.Event()

        sync.clear()

        class FailToRespond(Exception):

            pass

        class FailingDuringResponseHandler(TCPConnectionHandler):

            def handle_connection(self):

                req = self.rfile.readline()

                threading.currentThread().set_ready_event(sync)

                raise FailToRespond()

        server = self.get_server(

            connection_handler_class=FailingDuringResponseHandler)

        client = self.get_client()

        client.connect((server.host, server.port))

        client.write('ping\n')

        sync.wait()

        self.assertRaises(FailToRespond, server.pending_exception)

    def test_exception_swallowed_while_serving(self):

        sync = threading.Event()

        sync.clear()

        class CantServe(Exception):

            pass

        class FailingWhileServingConnectionHandler(TCPConnectionHandler):

            def handle(self):

                # We want to sync with the thread that is serving the

                # connection.

                threading.currentThread().set_ready_event(sync)

                raise CantServe()

        server = self.get_server(

            connection_handler_class=FailingWhileServingConnectionHandler)

        # Install the exception swallower

        server.set_ignored_exceptions(CantServe)

        client = self.get_client()

        # Connect to the server so the exception is raised there

        client.connect((server.host, server.port))

        # Wait for the exception to propagate.

        sync.wait()

        # The connection wasn't served properly but the exception should have

        # been swallowed.

        server.pending_exception()