KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use failure_counter method in Testify

Best Python code snippet using Testify_python

watchdog_server.py

Source:watchdog_server.py

1import socket2import logging3from threading import Thread, Lock4import os5class Watchdog_server:6    def __init__(self):7        # variables8        self.Psocket_host = "127.0.0.1"9        self.Vsocket_host = "127.0.0.1"10        self.Psocket_port = 1023711        self.Vsocket_port = 1023812        self.Pmutex = Lock()13        self.Vmutex = Lock()14        self.Pfail_flag = False15        self.Vfail_flag = False16        self.stop_flag = False17        18        # logging setup19        format = "%(asctime)s: %(message)s"20        logging.basicConfig(format=format, level=logging.DEBUG, datefmt="%H:%M:%S") 21    def start(self):22        # start theads23        self.P_thread = Thread(target=self.__Pwatchdog_start, daemon=True)24        # self.V_thread = Thread(target=self.__Vwatchdog_start, daemon=True)25        self.P_thread.start()26        # self.V_thread.start()27    def stop(self):28        self.stop_flag = True29        self.P_thread.join()30        # self.V_thread.join()31    32    def isPConnected(self):33        self.Pmutex.acquire()34        result = self.Pfail_flag35        self.Pmutex.release()36        return (not result)37    38    def isVConnected(self):39        self.Vmutex.acquire()40        result = self.Vfail_flag41        self.Vmutex.release()42        return (not result)43    def __Pwatchdog_start(self):44        self.Psocket = self.SocketHelper(self.Psocket_host, self.Psocket_port)45        logging.info("[P]Watchdog for physical board started")46        failure_counter = 047        # expect to receive 0 from Psocket48        # if both fail flags are false, reply 0, otherwise reply 149        while not self.stop_flag:50            try:51                result = self.Psocket.read_data()52                self.Psocket.set_timeout(3)53                if result == 1:54                    # fail to receive data, increment failure counter55                    failure_counter += 156                    if failure_counter >= 3:57                        self.Pmutex.acquire()58                        self.Pfail_flag = True59                        self.Pmutex.release()60                        break61                elif result == 0:62                        self.Vmutex.acquire()63                        if self.Vfail_flag:64                            # visual side disconnected65                            reply_msg = 166                        else:67                            # everything is normal68                            failure_counter = 069                            reply_msg = 070                        self.Vmutex.release()71                        self.Psocket.send_data(reply_msg)72                self.Vmutex.acquire()73                self.Pmutex.acquire()74                logging.info(f"[P]Received: {result} failure_counter: {failure_counter} Pflag: {self.Pfail_flag} Vflag: {self.Vfail_flag}")75                self.Pmutex.release()76                self.Vmutex.release()77            except Exception as e:78                logging.info("[P]"+str(e))79                # any error will conclude to Psocket failure80                self.Pmutex.acquire()81                self.Pfail_flag = True82                self.Pmutex.release()83                break84        85        logging.info("[P]Physical board connection lost")86        self.Psocket.close_socket()87        os.system("pkill python")88    89    def __Vwatchdog_start(self):90        self.Vsocket = self.SocketHelper(self.Vsocket_host, self.Vsocket_port)91        logging.info("[V]Watchdog for virtual board started")92        failure_counter = 093        # expect to receive 0 from Vsocket94        # if both fail flags are false, reply 0, otherwise reply 195        while not self.stop_flag:96            try:97                result = self.Vsocket.read_data()98                self.Vsocket.set_timeout(3)99                if result == 1:100                    # fail to receive data, increment failure counter101                    failure_counter += 1102                    if failure_counter >= 3:103                        self.Vmutex.acquire()104                        self.Vfail_flag = True105                        self.Vmutex.release()106                        break107                elif result == 0:108                        self.Pmutex.acquire()109                        if self.Pfail_flag:110                            # physical side disconnected111                            reply_msg = 1112                        else:113                            # everything is normal114                            failure_counter = 0115                            reply_msg = 0116                        self.Pmutex.release()117                        self.Vsocket.send_data(reply_msg)118                self.Pmutex.acquire()119                self.Vmutex.acquire()120                logging.info(f"[V]Received: {result} failure_counter: {failure_counter} Pflag: {self.Pfail_flag} Vflag: {self.Vfail_flag}")121                self.Vmutex.release()122                self.Pmutex.release()123            except Exception as e:124                logging.info("[V]"+str(e))125                # any error will conclude to Vsocket failure126                self.Vmutex.acquire()127                self.Vfail_flag = True128                self.Vmutex.release()129                break130        131        logging.info("[P]Virtual board connection lost")132        self.Vsocket.close_socket()133    class SocketHelper:134        def __init__(self, host, port):135            self.s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)136            self.s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)137            self.s.bind((host, port))138            self.s.listen(5)139            self.conn, self.addr = self.s.accept()140        141        def set_timeout(self, timeout):142            self.s.settimeout(timeout)143        144        def send_data(self, content):145            self.conn.send(bytes(str(content), "utf-8"))146        def read_data(self):147            try:148                buf = self.conn.recv(1024)149                return int(buf.decode("utf-8"))150            except:151                # 1 means connection fail152                return 1153        154        def close_socket(self):155            self.conn.close()...

API_tester.py

Source:API_tester.py

1import sys2import json3import utility4import generate_request5'''6Tool used to test API based on specified JSON.7Can be called from cli with jsonfile as parameter.8'''9def testing_tool(jsonfile = "output.json"):10    # variable used to store failed and passed messages11    results_list = []12    failure_counter = 013    # open json file to be tested14    with open(jsonfile) as f:15        data = json.load(f)16        x = 117        for i in data:18            print("\nTest", x) #counter19            20            # post request case21            if(i['type'].lower() == 'post'):22                # send post request23                post_req = generate_request.post_request(i['url'], i['body'], i['header'])24                # check_url = i['check']['url'].replace('{{id}}', i['body']['contact']['id'])25                get_req = generate_request.get_request(i['check']['url'], i['header'])26                # assert for response code and content27                try:28                    utility.request_assert(post_req.status_code, i['check']['success_code'])29                    utility.request_assert(get_req.status_code, 200)30                    utility.content_assert(i['body'], get_req)31                    results_list.append(utility.generate_pass_msg(x))32                    33                except Exception as e:34                    results_list.append(utility.generate_error_msg(x, e))35                    failure_counter += 136            # get request case37            elif(i['type'].lower() == 'get'):38                # send get request39                get_req = generate_request.get_request(i['url'], i['header'])40                # assert for response code41                try:42                    utility.request_assert(get_req.status_code, i['check']['success_code'])43                    results_list.append(utility.generate_pass_msg(x))44                    45                except Exception as e:46                    results_list.append(utility.generate_error_msg(x, e))47                    failure_counter += 148            # delete request case49            elif (i['type'].lower() == 'delete'):50                #send delete request51                del_req = generate_request.delete_request(i['url'], i['header'])52                get_req = generate_request.get_request(i['check']['url'], i['header'])53                try:54                    #assert for response code55                    utility.request_assert(del_req.status_code, i['check']['success_code'])56                    utility.request_assert(get_req.status_code, 404)57                    results_list.append(utility.generate_pass_msg(x))58                    59                except AssertionError as e:60                    results_list.append(utility.generate_error_msg(x, e))61                    failure_counter += 162                63            # put request case64            elif (i['type'].lower() == 'put'):65                # send put request66                put_req = generate_request.put_request(i['url'], i['body'], i['header'])67                get_req = generate_request.get_request(i['check']['url'], i['header'])68                69                try:70                    #assert for response code and content71                    utility.request_assert(put_req.status_code, i['check']['success_code'])72                    utility.request_assert(get_req.status_code, 200)73                    utility.content_assert(i['body'], get_req)74                    results_list.append(utility.generate_pass_msg(x))75                    76                except Exception as e:77                    results_list.append(utility.generate_error_msg(x,e))78                    failure_counter += 179                80            x += 181    print("\nTest Summary:")82    for i in results_list:83        print(i)84    success_rate = failure_counter/len(results_list)85    print("\nSuccess: "+str(len(results_list)-failure_counter)+"\nFailed: "+str(failure_counter))86    print("Success Rate: "+str(100 - (success_rate * 100))+str("%"))87# allows tool to function in cli by passing jsonfile argument88if len(sys.argv) > 1:...

test_server_down.py

Source:test_server_down.py

1#!/usr/bin/env python2import cmemcached3mc = cmemcached.Client(['127.0.0.1:11211', '127.0.0.1:11212', '127.0.0.1:11213', '127.0.0.1:11214'])4num = 100005keys = ['key%d' % k for k in xrange(num)]6success_counter = 07failure_counter = 08def print_counter():9    print 'success_counter is ', success_counter10    print 'failure_counter is ', failure_counter11    assert failure_counter+success_counter == num12    print 'mis rate is ', float(failure_counter)/num13while True:14    for key in keys:15        mc.set(key, 'aa')16    for key in keys:17        if mc.get(key) == 'aa':18            success_counter += 119        else:20            failure_counter += 121    print_counter()22    success_counter = 0...

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.