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