blob: 5555b625cb40c7c276105faab99fcedb2131bc58 [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
Patrick Williams57318182022-12-08 06:18:26 -06009from robot.libraries.BuiltIn import BuiltIn
George Keishinge635ddc2022-12-08 07:38:02 -060010from multiprocessing import Process, Manager
11import os
12import datetime
13import gen_print as gp
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.
50 for ix in range(int(num_process)):
George Keishinge635ddc2022-12-08 07:38:02 -060051 task = Process(target=execute_keyword,
52 args=(keyword_name, return_dict))
George Keishingd942b512017-06-06 14:34:55 -050053 process_list.append(task)
54 task.start()
55
56 # Wait for process to complete.
57 for task in process_list:
58 task.join()
59
60 # Return function return codes.
61 return return_dict
shrsuman1235fc20cb2021-02-02 04:55:47 -060062
63
64def execute_keyword_args(keyword_name, args, return_dict):
65 r"""
66 Execute a robot keyword with arguments.
67 In addition to running the caller's keyword, this function will:
68 - Add an entry to the return_dict
69 Description of argument(s):
70 keyword_name Keyword name to be executed.
71 args Arguments to keyword.
72 return_dict A dictionary consisting of pid/process status for the
73 keys/values. This function will append a new entry to
74 this dictionary.
75 """
76
77 execution_time = datetime.datetime.now()
78
79 status = BuiltIn().run_keyword_and_return_status(keyword_name, *args)
80
81 # Build execution time:<status> dictionary.
82 return_dict[str(execution_time)] = str(status)
83
84
85def execute_process_multi_keyword(number_args, *keyword_names):
86 r"""
87 Execute multiple robot keywords with arguments via multiprocessing process.
88
89 Description of argument(s):
90 number_args Number of argument in keywords.
91 keyword_names Keyword name to be executed.
92 """
93
94 manager = Manager()
95 return_dict = manager.dict()
96 process_list = []
97 # Append each keyword with its arguments in a process to execute.
98 for keywords_data in keyword_names:
99 keyword_args = tuple(keywords_data.split(" ")[-number_args:])
100 keyword_name = " ".join(keywords_data.split(" ")[:-number_args])
George Keishinge635ddc2022-12-08 07:38:02 -0600101 task = Process(target=execute_keyword_args,
102 args=(keyword_name, keyword_args, return_dict))
shrsuman1235fc20cb2021-02-02 04:55:47 -0600103 process_list.append(task)
104 task.start()
105
106 # Wait for process to complete.
107 for task in process_list:
108 task.join()
109 return return_dict
110
111
112def get_current_date_time():
113 r"""
114 Gets current time.
115 """
116
117 current_time = datetime.datetime.now().strftime("%H:%M:%S.%f")
118 return current_time