How to use kill_server method in Airtest

Best Python code snippet using Airtest

failure_test.py

Source:failure_test.py Github

copy

Full Screen

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__':...

Full Screen

Full Screen

app.py

Source:app.py Github

copy

Full Screen

...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 ###...

Full Screen

Full Screen

server.py

Source:server.py Github

copy

Full Screen

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

Full Screen

Full Screen

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run Airtest automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful