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
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