aboutsummaryrefslogtreecommitdiffstats
path: root/tools/teststepstools.py
blob: 33db8f796155b707eb03791602eebc6fb2d3892b (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
# Copyright 2016-2017 Intel Corporation.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Various helper functions for step driven testcases
"""

import logging
import subprocess
import locale
from tools.functions import filter_output
from tools.tasks import run_background_task

_LOGGER = logging.getLogger(__name__)

class TestStepsTools(object):
    """ Various tools and functions used by step driven testcases
    """
    # Functions use nonstandard names to avoid conflicts with
    # standard python keywords.
    # pylint: disable=invalid-name
    @staticmethod
    def Assert(condition):
        """ Evaluate given `condition' and raise AssertionError
            in case, that evaluation fails
        """
        try:
            assert TestStepsTools.Eval(condition)
        except AssertionError:
            _LOGGER.error('Condition %s is not True', condition)
            raise

        return True

    @staticmethod
    def validate_Assert(result, dummy_condition):
        """ Validate evaluation of given `condition'
        """
        return result

    @staticmethod
    def Eval(expression):
        """ Evaluate python `expression' and return its result
        """
        # pylint: disable=eval-used
        return eval(expression)

    @staticmethod
    def validate_Eval(result, dummy_expression):
        """ Validate result of python `expression' evaluation
        """
        return result is not None

    @staticmethod
    def Exec_Python(code):
        """ Execute a python `code' and return True on success
        """
        # pylint: disable=exec-used
        try:
            exec(code, globals())
        # pylint: disable=broad-except
        # pylint: disable=bare-except
        except:
            _LOGGER.error('Execution of following code has failed %s', code)
            return False
        return True

    @staticmethod
    def validate_Exec_Python(result, dummy_code):
        """ Validate result of python `code' execution
        """
        return result

    @staticmethod
    def Exec_Shell(command, regex=None):
        """ Execute a shell `command' and return its output filtered
            out by optional `regex' expression.
        """
        try:
            output = subprocess.check_output(command, shell=True)
        except OSError:
            return None

        output = output.decode(locale.getdefaultlocale()[1])

        if regex:
            return filter_output(output, regex)

        return output

    @staticmethod
    def validate_Exec_Shell(result, dummy_command, dummy_regex=None):
        """ validate result of shell `command' execution
        """
        return result is not None

    @staticmethod
    def Exec_Shell_Background(command):
        """ Execute a shell `command' at the background and return its PID id
        """
        try:
            pid = run_background_task(command.split(), _LOGGER, "Background task: {}".format(command))
            return pid
        except OSError:
            return None

    @staticmethod
    def validate_Exec_Shell_Background(result, dummy_command, dummy_regex=None):
        """ validate result of shell `command' execution on the background
        """
        return result is not None
:returns: True - in case that traffic generator should be executed False - if traffic generation is not required """ if self._mode == 'trafficgen-off': time.sleep(2) self._logger.debug("All is set. Please run traffic generator manually.") input(os.linesep + "Press Enter to terminate vswitchperf..." + os.linesep + os.linesep) return False elif self._mode == 'trafficgen-pause': time.sleep(2) while True: choice = input(os.linesep + 'Transmission paused, should' ' transmission be resumed? [y/n]' + os.linesep).lower() if choice in ('yes', 'y', 'ye'): return True elif choice in ('no', 'n'): self._logger.info("Traffic transmission will be skipped.") return False else: print("Please respond with 'yes', 'y', 'no' or 'n' ", end='') return True def send_traffic(self, traffic): """Triggers traffic to be sent from the traffic generator. This is a blocking function. :param traffic: A dictionary describing the traffic to send. """ self._logger.debug('send_traffic with ' + str(self._traffic_gen_class)) self.configure(traffic) def send_traffic_async(self, traffic, dummy_function): """Triggers traffic to be sent asynchronously. This is not a blocking function. :param traffic: A dictionary describing the traffic to send. :param function: A dictionary describing the function to call between send and wait in the form: function = { 'function' : package.module.function, 'args' : args } If this function requires more than one argument, all should be should be passed using the args list and appropriately handled. """ self._logger.debug('send_traffic_async with ' + str(self._traffic_gen_class)) self.configure(traffic) def stop_traffic(self): """Kills traffic being sent from the traffic generator. """ self._logger.debug("stop_traffic()") def print_results(self): """IResult interface implementation. """ counter = 0 for item in self._results: logging.info("Record: " + str(counter)) counter += 1 for(key, value) in list(item.items()): logging.info(" Key: " + str(key) + ", Value: " + str(value)) def get_results(self): """IResult interface implementation. """ return self._results def validate_send_traffic(self, dummy_result, dummy_traffic): """Verify that send traffic has succeeded """ if self._results: if 'b2b_frames' in self._results[-1]: return float(self._results[-1]['b2b_frames']) > 0 elif 'throughput_rx_fps' in self._results[-1]: return float(self._results[-1]['throughput_rx_fps']) > 0 else: return True else: return False def validate_get_results(self, result): """Verify that results has been returned """ return self._results == result