1#!/usr/bin/env python3 2 3r""" 4This module contains keyword functions to support multiprocessing 5execution of keywords where generic robot keywords don't support. 6 7""" 8 9import datetime 10import os 11from multiprocessing import Manager, Process 12 13import gen_print as gp 14from robot.libraries.BuiltIn import BuiltIn 15 16 17def execute_keyword(keyword_name, return_dict): 18 r""" 19 Execute a robot keyword. 20 In addition to running the caller's keyword, this function will: 21 - Add an entry to the return_dict 22 23 Description of argument(s): 24 keyword_name Keyword name to be executed. 25 return_dict A dictionary consisting of pid/process status for the 26 keys/values. This function will append a new entry to 27 this dictionary. 28 """ 29 30 pid = os.getpid() 31 status = BuiltIn().run_keyword_and_return_status(keyword_name) 32 33 # Build PID:<status> dictionary. 34 return_dict[str(pid)] = str(status) 35 36 37def execute_process(num_process, keyword_name): 38 r""" 39 Execute a robot keyword via multiprocessing process. 40 41 Description of argument(s): 42 num_process Number of times keyword to be executed. 43 keyword_name Keyword name to be executed. 44 """ 45 46 manager = Manager() 47 return_dict = manager.dict() 48 process_list = [] 49 50 # Append user-defined times process needed to execute. 51 for ix in range(int(num_process)): 52 task = Process( 53 target=execute_keyword, args=(keyword_name, return_dict) 54 ) 55 process_list.append(task) 56 task.start() 57 58 # Wait for process to complete. 59 for task in process_list: 60 task.join() 61 62 # Return function return codes. 63 return return_dict 64 65 66def execute_keyword_args(keyword_name, args, return_dict): 67 r""" 68 Execute a robot keyword with arguments. 69 In addition to running the caller's keyword, this function will: 70 - Add an entry to the return_dict 71 Description of argument(s): 72 keyword_name Keyword name to be executed. 73 args Arguments to keyword. 74 return_dict A dictionary consisting of pid/process status for the 75 keys/values. This function will append a new entry to 76 this dictionary. 77 """ 78 79 execution_time = datetime.datetime.now() 80 81 status = BuiltIn().run_keyword_and_return_status(keyword_name, *args) 82 83 # Build execution time:<status> dictionary. 84 return_dict[str(execution_time)] = str(status) 85 86 87def execute_process_multi_keyword(number_args, *keyword_names): 88 r""" 89 Execute multiple robot keywords with arguments via multiprocessing process. 90 91 Description of argument(s): 92 number_args Number of argument in keywords. 93 keyword_names Keyword name to be executed. 94 """ 95 96 manager = Manager() 97 return_dict = manager.dict() 98 process_list = [] 99 # Append each keyword with its arguments in a process to execute. 100 for keywords_data in keyword_names: 101 keyword_args = tuple(keywords_data.split(" ")[-number_args:]) 102 keyword_name = " ".join(keywords_data.split(" ")[:-number_args]) 103 task = Process( 104 target=execute_keyword_args, 105 args=(keyword_name, keyword_args, return_dict), 106 ) 107 process_list.append(task) 108 task.start() 109 110 # Wait for process to complete. 111 for task in process_list: 112 task.join() 113 return return_dict 114 115 116def get_current_date_time(): 117 r""" 118 Gets current time. 119 """ 120 121 current_time = datetime.datetime.now().strftime("%H:%M:%S.%f") 122 return current_time 123