How to use kill_server method in Airtest

1#!/usr/bin/python2#########################################################################3# failure_test.py4#########################################################################5import server6import sys7import os8import unittest9import command10import pickle11import socket12import subprocess13import time14import message15from multiprocessing import Queue, Process, Lock16import random17from lock_file import *18import client19class failure_test(unittest.TestCase):20    ###############################################################21    # Bring up a full Paxos group with multiple server instances22    ###############################################################23    def setUp(self):24        # set test server size25        self.TOTAL_SERVERS = 526        # initialize server list27        self.server_list = []28        # generate the host numbers and ports of server connections29        for i in range(9000, 9000 + 2 * self.TOTAL_SERVERS, 2):30            server_entry = dict()31            server_entry["host"] = "localhost"32            server_entry["internal_port"] = i + 133            server_entry["client_port"] = i34            self.server_list.append(server_entry)35        # bring up each server36        self.servers = []37        # iterate through address list and instantiate servers38        for i in range(0, len(self.server_list)):39            # instantiate new servers40            new_server = server.PAXOS_member(i, self.server_list)41            self.servers.append(new_server)42        assert(len(self.servers) == len(self.server_list))43        assert(len(self.servers) == self.TOTAL_SERVERS)44        # initialize new servers45        for i in range(0, len(self.servers)):46            s = self.server_list[i]47            assert s48            self.servers[i].initialize_paxos()49    ###############################################################50    # Shutdown the Paxos group51    ###############################################################52    def tearDown(self):53        # directly inject an exit message to each of the queues54        exit_msg = message.message(message.MESSAGE_TYPE.EXIT,55                                   None, None, None, 'localhost', 9000, None)56 57        for s in self.servers:58            s.acceptor_queue.put(exit_msg)59            s.proposer_queue.put(exit_msg)60        # join the acceptor and proposer processes for each server61        for s in self.servers:62            s.listening_process.terminate()63            s.listening_process.join(5)64            s.acceptor_process.join(5)65            s.proposer_process.join(5)66        # do a manual clean up of the leftover sockets67        dump = os.popen("lsof -n -i | grep python").read()68        lines = dumps.split("\n")69        for line in lines:70            fields = line.split()71            pid = fields[1]72            os.system("kill " + pid)73        # a manual validation of this test is necessary due to the non-determinism of server failures74    ##########################################################75    # Test if Paxos group completes even with single node76    #  experiencing a failure77    # Do not attempt recovery78    ##########################################################79    def test_single_server_failure(self):80        print "\n[Info] ##########[INTEGRATION TEST MULTIPLE CLIENT MULTIPLE SERVER TEST]########## \n\n"81        LOCKS = 10082        kill_server = self.servers[random.randint(0, len(self.servers)-1)]83        # generate the lock files84        for i in range(0, len(self.server_list)):85            filename = "client_" + str(i) + ".txt"86            make_simple_file(LOCKS, filename)87        # instantiate a client to each server88        client_list = []89        for i in range(0, len(self.server_list)):90            port = self.server_list[i]["client_port"]91            host = self.server_list[i]["host"]92            assert((int(port) % 2) == 0)93            cli = client.client(94                "client_" + str(i) + ".txt", host, port, len(client_list))95            client_list.append(cli)96        time.sleep(1)97        # implement a hard terminate on each server98        kill_server.proposer_process.terminate()99        kill_server.acceptor_process.terminate()100        kill_server.listening_process.terminate()101        try:102            kill_server.listening_process.join(1)103        except:104            pass105        try:106            kill_server.acceptor_process.join(1)107        except:108            pass109        try:110            kill_server.proposer_process.join(1)111        except:112            pass113        self.servers.remove(kill_server)114        # join each client115        failed = False116        for c in client_list:117            try:118                c.c_process.join()119            except Exception, e:120                c.terminate()121                failed = True122        assert(not failed)123    ##########################################################124    # Test if Paxos group completes even with slightly less 125    # than a majority of nodes experiencing a failure126    # Do not attempt recovery127    ##########################################################128    def test_multi_server_failure(self):129        print "\n[Info] ##########[INTEGRATION TEST MULTIPLE CLIENT MULTIPLE SERVER TEST]########## \n\n"130        LOCKS = 100131        # generate the lock files132        for i in range(0, len(self.server_list)):133            filename = "client_" + str(i) + ".txt"134            make_simple_file(LOCKS, filename)135        # instantiate a client to each server136        client_list = []137        for i in range(0, len(self.server_list)):138            port = self.server_list[i]["client_port"]139            host = self.server_list[i]["host"]140            assert((int(port) % 2) == 0)141            cli = client.client(142                "client_" + str(i) + ".txt", host, port, len(client_list))143            client_list.append(cli)144        time.sleep(1)145        for i in range(0, len(self.server_list)/2  - 1 + (len(self.server_list) % 2)):146            kill_server = self.servers[i]147            kill_server.proposer_process.terminate()148            kill_server.acceptor_process.terminate()149            kill_server.listening_process.terminate()150            try:151                kill_server.listening_process.join(1)152                print "[Info] Terminated server " + str(i) + "..."153            except:154                pass155            self.server.remove(kill_server)156        # join each client157        failed = False158        for c in client_list:159            try:160                c.c_process.join()161            except Exception, e:162                c.terminate()163                failed = True164        assert(not failed)165if __name__ == '__main__':...

...17define("debug", default=True, help="run in debug mode", type=bool)18define("id", default=0, help="instance id used for multiple processes", type=int)19# Obtain the global Logger20_logger = logging.getLogger(__name__)21def kill_server(sig, frame):22    print("Received signal=%d, quit!" % sig)23    tornado.ioloop.IOLoop.instance().stop()24def main():25    # signal.signal(signal.SIGPIPE, signal.SIG_IGN)26    signal.signal(signal.SIGINT, kill_server)27    # signal.signal(signal.SIGQUIT, kill_server)28    signal.signal(signal.SIGTERM, kill_server)29    # signal.signal(signal.SIGHUP, kill_server)30    ###31    tornado.options.parse_command_line()32    formatter = logging.Formatter("[%(asctime)s] %(levelname)s (%(filename)s:%(lineno)d) %(message)s")33    for logHandler in logging.getLogger().handlers:34        logHandler.setFormatter(formatter)35    ###...

...22    fmt = '%(asctime)s [%(levelname)s %(name)s pid:%(process)d port:{port} %(filename)s:%(lineno)d] %(message)s'.format(23            port=options.port or '')24    formatter = LogFormatter(color=True, fmt=fmt, datefmt=datefmt)25    access_log.handlers[0].setFormatter(formatter)26def kill_server(sig, frame):27    logging.info('Caught signal: %s stop server(%s)' % (sig, options.port))28    tornado.ioloop.IOLoop.instance().stop()29def register_signal():30    signal.signal(signal.SIGPIPE, signal.SIG_IGN)31    signal.signal(signal.SIGINT, kill_server)32    signal.signal(signal.SIGQUIT, kill_server)33    # signal.signal(signal.SIGTERM, kill_server)34    signal.signal(signal.SIGHUP, kill_server)35def run_server():36    server = tornado.httpserver.HTTPServer(application, xheaders=True, no_keep_alive=False)37    server.listen(options.port)38    tornado.ioloop.IOLoop.instance().start()39register_signal()40init_logging()...