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

How to use kill_process_tree method in localstack

Best Python code snippet using localstack_python

honeypot.py

Source:honeypot.py

...23                for p in psutil.process_iter():24                    if p.open_files() != []:25                        if event.src_path in p.open_files()[0][0]:26                            print('find danger')27                            kill_process_tree(p.pid,signal.SIGKILL)28            print(entropy_dict[event.src_path])29    def on_created(self, event):30        if event.is_directory:31            print('---------------------')32            print('created:',event.src_path)33            print(entropy_dict[event.src_path])34            danger = detecter.detect(event.src_path, entropy_dict)35            print(danger)36            if danger > 0:37                for p in psutil.process_iter():38                    if p.open_files() != []:39                        if event.src_path in p.open_files()[0][0]:40                            print('find danger')41                            kill_process_tree(p.pid,signal.SIGKILL)42            print(entropy_dict[event.src_path])43    def on_deleted(self, event):44        if event.is_directory:45            print('---------------------')46            print('deleted:',event.src_path)47            print(entropy_dict[event.src_path])48            danger = detecter.detect(event.src_path, entropy_dict)49            print(danger)50            if danger > 0:51                for p in psutil.process_iter():52                    if p.open_files() != []:53                        if event.src_path in p.open_files()[0][0]:54                            print('find danger')55                            kill_process_tree(p.pid,signal.SIGKILL)56            print(entropy_dict[event.src_path])57    def on_modified(self, event):58        if event.is_directory:59            print('---------------------')60            print('modified',event.src_path)61            print(entropy_dict[event.src_path])62            danger = detecter.detect(event.src_path, entropy_dict)63            print(danger)64            if danger > 0:65                for p in psutil.process_iter():66                    if p.open_files() != []:67                        if event.src_path in p.open_files()[0][0]:68                            print('find danger')69                            kill_process_tree(p.pid,signal.SIGKILL)70            print(entropy_dict[event.src_path])71"""72ä½¿ç¨watchdog çæ§æä»¶çåå73"""74def watching(directory,Deside):75    detecter.entropy_init(entropy_dict)76    print('initial entropy is:', entropy_dict, '\n')77    # åå»ºè§å¯èå¯¹è±¡78    observer = Observer()79    # åå»ºäºä»¶å¤çå¯¹è±¡80    fileHandler = MyDirEventHandler()81    # ä¸ºè§å¯èè®¾ç½®è§å¯å¯¹è±¡ä¸å¤çäºä»¶å¯¹è±¡82    observer.schedule(fileHandler, directory, True)83    observer.start()...

signal_utils.py

Source:signal_utils.py

...36    terminate_parent_process(proc)37  elif terminate_strategy == TerminateStrategy.KILL_PROCESS_GROUP:38    killpg_gracefully(proc)39  elif terminate_strategy == TerminateStrategy.KILL_PROCESS_TREE:40    kill_process_tree(proc)41  else:42    raise Fail("Invalid timeout_kill_strategy = '{0}'. Use TerminateStrategy class constants as a value.".format(terminate_strategy))43def killpg_gracefully(proc, timeout=GRACEFUL_PG_KILL_TIMEOUT_SECONDS):44  """45  Tries to kill pgroup (process group) of process with SIGTERM.46  If the process is still alive after waiting for timeout, SIGKILL is sent to the pgroup.47  """48  from resource_management.core import sudo49  from resource_management.core.logger import Logger50  if proc.poll() == None:51    try:52      pgid = os.getpgid(proc.pid)53      sudo.kill(-pgid, signal.SIGTERM)54      for i in xrange(10*timeout):55        if proc.poll() is not None:56          break57        time.sleep(0.1)58      else:59        Logger.info("Cannot gracefully kill process group {0}. Resorting to SIGKILL.".format(pgid))60        sudo.kill(-pgid, signal.SIGKILL)61        proc.wait()62    # catch race condition if proc already dead63    except OSError:64      pass65    66def terminate_parent_process(proc):67  if proc.poll() == None:68    try:69      proc.terminate()70      proc.wait()71    # catch race condition if proc already dead72    except OSError:73      pass74    75def kill_process_tree(proc):76  from resource_management.core import shell77  current_directory = os.path.dirname(os.path.abspath(__file__))78  kill_tree_script = "{0}/files/killtree.sh".format(current_directory)79  if proc.poll() == None:80    shell.checked_call(["bash", kill_tree_script, str(proc.pid), str(signal.SIGKILL)])...

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.