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

How to use kill_monitor method in autotest

Best Python code snippet using autotest_python

monitor_db_babysitter

Source:monitor_db_babysitter

...34        command_output = cmd_out.stdout + cmd_out.stderr35    except error.CmdError:36        command_output = 'Timed out'37    return banner_output % command_output38def kill_monitor():39    logging.info("Killing monitor_db")40    # try shutdown first41    utils.signal_program(monitor_db.PID_FILE_PREFIX, sig=signal.SIGINT)42    if utils.program_is_alive(monitor_db.PID_FILE_PREFIX): # was it killed?43        # give it some time to shutdown44        time.sleep(30)45        # kill it46        utils.signal_process(monitor_db.PID_FILE_PREFIX)47def handle_sigterm(signum, frame):48    logging.info('Caught SIGTERM')49    kill_monitor()50    utils.delete_pid_file_if_exists(monitor_db.BABYSITTER_PID_FILE_PREFIX)51    sys.exit(1)52signal.signal(signal.SIGTERM, handle_sigterm)53SiteMonitorProc = utils.import_site_class(54    __file__, 'autotest_lib.scheduler.site_monitor_db_babysitter',55    'SiteMonitorProc', object)56class MonitorProc(SiteMonitorProc):57    def __init__(self, do_recovery=False):58        args = [monitor_db_path]59        if do_recovery:60            args.append("--recover-hosts")61        args.append(results_dir)62        kill_monitor()63        environ = os.environ64        scheduler_config = scheduler_logging_config.SchedulerLoggingConfig65        log_name = scheduler_config.get_log_name()66        os.environ['AUTOTEST_SCHEDULER_LOG_NAME'] = log_name67        scheduler_log_dir = scheduler_config.get_server_log_dir()68        self.log_path = os.path.join(scheduler_log_dir, log_name)69        self.log_size = 070        self.last_log_change = time.time()71        logging.info("STARTING monitor_db with log file %s" % self.log_path)72        self.args = args73        # Allow site specific code to run, set environment variables and74        # modify self.args if desired.75        super(MonitorProc, self).__init__()76    def start(self):...

test_multiprocessing_value.py

Source:test_multiprocessing_value.py

1import logging2from multiprocessing import Process, Value3import os4import sys5import time6logging.basicConfig()7logger = logging.getLogger(__name__)8logger.setLevel(logging.DEBUG)9class Monitor:10    def __init__(self):11        self.monitor_active = False12    def start(self, monitor_state, kill_monitor):13        Monitor.monitor(self, monitor_state, kill_monitor)14    def stop(self):15        pass16    def monitor(self, monitor_state, kill_monitor):#, monitor_active):17        self.monitor_active = True18        loop_start = time.time()19        while (True):20            try:21                if monitor_state.value == 0:22                    monitor_state.value = 123                elif monitor_state.value == 1:24                    monitor_state.value = 025                if kill_monitor == 1 or (time.time() - loop_start) > 60:26                    break27                time.sleep(5)28            except Exception as e:29                logger.exception(e)30            except KeyboardInterrupt:31                logger.info('Exit signal received.')32                break33            finally:34                self.monitor_active = False35if __name__ == '__main__':36    try:37        monitor_state = Value('b', 0)38        kill_monitor = Value('b', 0)39        monitor = Monitor()40        arguments = tuple()41        keyword_arguments = {'monitor_state': monitor_state, 'kill_monitor': kill_monitor}42        monitor_proc = Process(target=monitor.start, args=arguments, kwargs=keyword_arguments)43        logger.info('Starting process.')44        monitor_proc.start()45        loop_start = time.time()46        while (True):47            logger.info('monitor_state.value: ' + str(monitor_state.value))48            time.sleep(5)49        #logger.info('Joining process.')50        #monitor_proc.join()51    except Exception as e:52        logger.exception('Unhandled exception in heartbeatmonitor.monitor.')53        logger.exception(e)54    except KeyboardInterrupt:55        logger.info('Exit signal received.')56    finally:57        monitor_proc.terminate()58        monitor_proc.join()...

stop_carrier.py

Source:stop_carrier.py

1# encoding:utf-82from monitor import monitor3import sys, settings, time, re4class kill_monitor(monitor):56    def gen_kill_cmd(self, spiders):7        """8        å®å¨å°ææ»æä¸ªç¬è«ï¼ä¸åå¯å¨9        :param spiders: get stateséè·åå°çç¬è«åååç¬¦ä¸²åè¡¨10        :return:11        """12        datas = []13        hosts = set()14        end = re.compile(r'-proxy$')15        for spider in spiders:16            host, carrier, number = spider.split(':')17            device = end.sub('', host)18            key = '%s -a host_name=%s -a num=%s' % (carrier.lower(), host, number)19            if device not in hosts:20                cmd_kill = "ps -ef | grep '%s_man.py' | grep -v grep | cut -c 9-15 | xargs kill -9" % carrier.lower()21                hosts.add(device)22                data_kill = {'cmd': cmd_kill, 'devices': [device]}23                datas.append(data_kill)24            cmd = "ps -ef | grep '%s' | grep -v grep | cut -c 9-15 | xargs kill -9" % key25            data = {'cmd': cmd, 'devices': [device]}26            datas.append(data)27        downloads_cmd = '~/spider/update_and_restart_all/download.sh'28        # downloads_cmd = '~/spider/update_and_restart_all/restart.sh'29        datas.insert(0, {'cmd': downloads_cmd, 'devices': list(hosts)})30        print(datas)31        self.post_cmd(datas)3233343536if __name__ == '__main__' :37    mon = kill_monitor()38    carrier = sys.argv[1].upper() or False39    if carrier:40        content = mon.get_content()41        all_spiders = mon.get_all_spider(content)42        spiders = mon.get_spider_by_name(all_spiders, carrier)
...

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.