blob: 2c214df3c48dbfb3b039a481f63d78f10c6640fb [file] [log] [blame]
George Keishinge7e91712021-09-03 11:28:44 -05001#!/usr/bin/env python3
George Keishingd942b512017-06-06 14:34:55 -05002
3r"""
4This module contains keyword functions to support multiprocessing
5execution of keywords where generic robot keywords don't support.
6
7"""
8
George Keishinge635ddc2022-12-08 07:38:02 -06009import datetime
Patrick Williams20f38712022-12-08 06:18:26 -060010import os
11from multiprocessing import Manager, Process
12
Patrick Williams20f38712022-12-08 06:18:26 -060013from robot.libraries.BuiltIn import BuiltIn
George Keishingd942b512017-06-06 14:34:55 -050014
15
George Keishingd942b512017-06-06 14:34:55 -050016def 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.
George Keishing84d089f2017-07-17 09:33:01 -050024 return_dict A dictionary consisting of pid/process status for the
George Keishingd942b512017-06-06 14:34:55 -050025 keys/values. This function will append a new entry to
26 this dictionary.
27 """
28
29 pid = os.getpid()
George Keishing84d089f2017-07-17 09:33:01 -050030 status = BuiltIn().run_keyword_and_return_status(keyword_name)
George Keishingd942b512017-06-06 14:34:55 -050031
George Keishing84d089f2017-07-17 09:33:01 -050032 # Build PID:<status> dictionary.
33 return_dict[str(pid)] = str(status)
George Keishingd942b512017-06-06 14:34:55 -050034
George Keishingd942b512017-06-06 14:34:55 -050035
George Keishingd942b512017-06-06 14:34:55 -050036def 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.
Sandhya Somashekar3f7ef6d2023-02-13 04:03:38 -060050 for _ix in range(int(num_process)):
Patrick Williams20f38712022-12-08 06:18:26 -060051 task = Process(
52 target=execute_keyword, args=(keyword_name, return_dict)
53 )
George Keishingd942b512017-06-06 14:34:55 -050054 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
shrsuman1235fc20cb2021-02-02 04:55:47 -060063
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])
Patrick Williams20f38712022-12-08 06:18:26 -0600102 task = Process(
103 target=execute_keyword_args,
104 args=(keyword_name, keyword_args, return_dict),
105 )
shrsuman1235fc20cb2021-02-02 04:55:47 -0600106 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
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