Fix pylint warnings in python libraries
[csit.git] / resources / libraries / python / VatExecutor.py
1 # Copyright (c) 2016 Cisco and/or its affiliates.
2 # Licensed under the Apache License, Version 2.0 (the "License");
3 # you may not use this file except in compliance with the License.
4 # You may obtain a copy of the License at:
5 #
6 #     http://www.apache.org/licenses/LICENSE-2.0
7 #
8 # Unless required by applicable law or agreed to in writing, software
9 # distributed under the License is distributed on an "AS IS" BASIS,
10 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
11 # See the License for the specific language governing permissions and
12 # limitations under the License.
13
14 """VAT executor library."""
15
16 import json
17
18 from robot.api import logger
19
20 from resources.libraries.python.ssh import SSH
21 from resources.libraries.python.constants import Constants
22
23
24 __all__ = ['VatExecutor']
25
26
27 def cleanup_vat_json_output(json_output):
28     """Return VAT JSON output cleaned from VAT clutter.
29
30     Clean up VAT JSON output from clutter like vat# prompts and such.
31
32     :param json_output: Cluttered JSON output.
33     :return: Cleaned up output JSON string.
34     """
35
36     retval = json_output
37     clutter = ['vat#', 'dump_interface_table error: Misc']
38     for garbage in clutter:
39         retval = retval.replace(garbage, '')
40     return retval
41
42
43 class VatExecutor(object):
44     """Contains methods for executing VAT commands on DUTs."""
45     def __init__(self):
46         self._stdout = None
47         self._stderr = None
48         self._ret_code = None
49
50     def execute_script(self, vat_name, node, timeout=10, json_out=True):
51         """Copy local_path script to node, execute it and return result.
52
53         :param vat_name: Name of the vat script file. Only the file name of
54         the script is required, the resources path is prepended automatically.
55         :param node: Node to execute the VAT script on.
56         :param timeout: Seconds to allow the script to run.
57         :param json_out: Require JSON output.
58         :return: (rc, stdout, stderr) tuple.
59         """
60
61         ssh = SSH()
62         ssh.connect(node)
63
64         remote_file_path = '{0}/{1}/{2}'.format(Constants.REMOTE_FW_DIR,
65                                                 Constants.RESOURCES_TPL_VAT,
66                                                 vat_name)
67         # TODO this overwrites the output if the vat script has been used twice
68         # remote_file_out = remote_file_path + ".out"
69
70         cmd = "sudo -S {vat} {json} < {input}".format(
71             vat=Constants.VAT_BIN_NAME,
72             json="json" if json_out is True else "",
73             input=remote_file_path)
74         (ret_code, stdout, stderr) = ssh.exec_command(cmd, timeout)
75         self._ret_code = ret_code
76         self._stdout = stdout
77         self._stderr = stderr
78
79         logger.trace("Command '{0}' returned {1}'".format(cmd, self._ret_code))
80         logger.trace("stdout: '{0}'".format(self._stdout))
81         logger.trace("stderr: '{0}'".format(self._stderr))
82
83         # TODO: download vpp_api_test output file
84         # self._delete_files(node, remote_file_path, remote_file_out)
85
86     def execute_script_json_out(self, vat_name, node, timeout=10):
87         """Pass all arguments to 'execute_script' method, then cleanup returned
88         json output."""
89         self.execute_script(vat_name, node, timeout, json_out=True)
90         self._stdout = cleanup_vat_json_output(self._stdout)
91
92     @staticmethod
93     def _delete_files(node, *files):
94         """Use SSH to delete the specified files on node.
95
96         :param node: Node in topology.
97         :param files: Files to delete.
98         :type node: dict
99         :type files: iterable
100         """
101
102         ssh = SSH()
103         ssh.connect(node)
104         files = " ".join([str(x) for x in files])
105         ssh.exec_command("rm {0}".format(files))
106
107     def script_should_have_failed(self):
108         """Read return code from last executed script and raise exception if the
109         script didn't fail."""
110         if self._ret_code is None:
111             raise Exception("First execute the script!")
112         if self._ret_code == 0:
113             raise AssertionError(
114                 "Script execution passed, but failure was expected")
115
116     def script_should_have_passed(self):
117         """Read return code from last executed script and raise exception if the
118         script failed."""
119         if self._ret_code is None:
120             raise Exception("First execute the script!")
121         if self._ret_code != 0:
122             raise AssertionError(
123                 "Script execution failed, but success was expected")
124
125     def get_script_stdout(self):
126         """Returns value of stdout from last executed script."""
127         return self._stdout
128
129     def get_script_stderr(self):
130         """Returns value of stderr from last executed script."""
131         return self._stderr
132
133     @staticmethod
134     def cmd_from_template(node, vat_template_file, **vat_args):
135         """Execute VAT script on specified node. This method supports
136         script templates with parameters.
137
138         :param node: Node in topology on witch the script is executed.
139         :param vat_template_file: Template file of VAT script.
140         :param vat_args: Arguments to the template file.
141         :return: List of JSON objects returned by VAT.
142         """
143         with VatTerminal(node) as vat:
144             return vat.vat_terminal_exec_cmd_from_template(vat_template_file,
145                                                            **vat_args)
146
147
148 class VatTerminal(object):
149     """VAT interactive terminal.
150
151     :param node: Node to open VAT terminal on.
152     :param json_param: Defines if outputs from VAT are in JSON format.
153     Default is True.
154     :type node: dict
155     :type json_param: bool
156
157     """
158
159     __VAT_PROMPT = "vat# "
160     __LINUX_PROMPT = ":~$ "
161
162     def __init__(self, node, json_param=True):
163         json_text = ' json' if json_param else ''
164         self.json = json_param
165         self._ssh = SSH()
166         self._ssh.connect(node)
167         self._tty = self._ssh.interactive_terminal_open()
168         self._ssh.interactive_terminal_exec_command(
169             self._tty,
170             'sudo -S {}{}'.format(Constants.VAT_BIN_NAME, json_text),
171             self.__VAT_PROMPT)
172         self._exec_failure = False
173
174     def __enter__(self):
175         return self
176
177     def __exit__(self, exc_type, exc_val, exc_tb):
178         self.vat_terminal_close()
179
180     def vat_terminal_exec_cmd(self, cmd):
181         """Execute command on the opened VAT terminal.
182
183         :param cmd: Command to be executed.
184
185         :return: Command output in python representation of JSON format or
186         None if not in JSON mode.
187         """
188         logger.debug("Executing command in VAT terminal: {}".format(cmd))
189         try:
190             out = self._ssh.interactive_terminal_exec_command(self._tty,
191                                                           cmd,
192                                                           self.__VAT_PROMPT)
193         except:
194             self._exec_failure = True
195             raise
196
197         logger.debug("VAT output: {}".format(out))
198         if self.json:
199             obj_start = out.find('{')
200             obj_end = out.rfind('}')
201             array_start = out.find('[')
202             array_end = out.rfind(']')
203
204             if -1 == obj_start and -1 == array_start:
205                 raise RuntimeError("VAT: no JSON data.")
206
207             if obj_start < array_start or -1 == array_start:
208                 start = obj_start
209                 end = obj_end + 1
210             else:
211                 start = array_start
212                 end = array_end + 1
213             out = out[start:end]
214             json_out = json.loads(out)
215             return json_out
216         else:
217             return None
218
219     def vat_terminal_close(self):
220         """Close VAT terminal."""
221         #interactive terminal is dead, we only need to close session
222         if not self._exec_failure:
223             self._ssh.interactive_terminal_exec_command(self._tty,
224                                                         'quit',
225                                                         self.__LINUX_PROMPT)
226         self._ssh.interactive_terminal_close(self._tty)
227
228     def vat_terminal_exec_cmd_from_template(self, vat_template_file, **args):
229         """Execute VAT script from a file.
230
231         :param vat_template_file: Template file name of a VAT script.
232         :param args: Dictionary of parameters for VAT script.
233         :return: List of JSON objects returned by VAT.
234         """
235         file_path = '{}/{}'.format(Constants.RESOURCES_TPL_VAT,
236                                    vat_template_file)
237         with open(file_path, 'r') as template_file:
238             cmd_template = template_file.readlines()
239         ret = []
240         for line_tmpl in cmd_template:
241             vat_cmd = line_tmpl.format(**args)
242             ret.append(self.vat_terminal_exec_cmd(vat_cmd.replace('\n', '')))
243         return ret

©2016 FD.io a Linux Foundation Collaborative Project. All Rights Reserved.
Linux Foundation is a registered trademark of The Linux Foundation. Linux is a registered trademark of Linus Torvalds.
Please see our privacy policy and terms of use.