diff options
Diffstat (limited to 'testcases/testcase.py')
| -rw-r--r-- | testcases/testcase.py | 114 |
1 files changed, 74 insertions, 40 deletions
diff --git a/testcases/testcase.py b/testcases/testcase.py index 3d9ffe4b..37cdefa6 100644 --- a/testcases/testcase.py +++ b/testcases/testcase.py @@ -68,7 +68,6 @@ class TestCase(object): self._loadgen = None self._output_file = None self._tc_results = None - self._settings_original = {} self._settings_paths_modified = False self._testcast_run_time = None self._versions = [] @@ -76,21 +75,18 @@ class TestCase(object): self._step_check = False # by default don't check result for step driven testcases self._step_vnf_list = {} self._step_result = [] + self._step_result_mapping = {} self._step_status = None + self._step_send_traffic = False # indication if send_traffic was called within test steps self._testcase_run_time = None - # store all GUEST_ specific settings to keep original values before their expansion - for key in S.__dict__: - if key.startswith('GUEST_'): - self._settings_original[key] = S.getValue(key) - - self._update_settings('VSWITCH', cfg.get('vSwitch', S.getValue('VSWITCH'))) - self._update_settings('VNF', cfg.get('VNF', S.getValue('VNF'))) - self._update_settings('TRAFFICGEN', cfg.get('Trafficgen', S.getValue('TRAFFICGEN'))) + S.setValue('VSWITCH', cfg.get('vSwitch', S.getValue('VSWITCH'))) + S.setValue('VNF', cfg.get('VNF', S.getValue('VNF'))) + S.setValue('TRAFFICGEN', cfg.get('Trafficgen', S.getValue('TRAFFICGEN'))) test_params = copy.deepcopy(S.getValue('TEST_PARAMS')) tc_test_params = cfg.get('Parameters', S.getValue('TEST_PARAMS')) test_params = merge_spec(test_params, tc_test_params) - self._update_settings('TEST_PARAMS', test_params) + S.setValue('TEST_PARAMS', test_params) S.check_test_params() # override all redefined GUEST_ values to have them expanded correctly @@ -109,6 +105,15 @@ class TestCase(object): self.desc = cfg.get('Description', 'No description given.') self.test = cfg.get('TestSteps', None) + # log testcase name and details + tmp_desc = functions.format_description(self.desc, 50) + self._logger.info('############################################################') + self._logger.info('# Test: %s', self.name) + self._logger.info('# Details: %s', tmp_desc[0]) + for i in range(1, len(tmp_desc)): + self._logger.info('# %s', tmp_desc[i]) + self._logger.info('############################################################') + bidirectional = S.getValue('TRAFFIC')['bidir'] if not isinstance(S.getValue('TRAFFIC')['bidir'], str): raise TypeError( @@ -162,6 +167,7 @@ class TestCase(object): self._traffic['l2'] = S.getValue(self._tunnel_type.upper() + '_FRAME_L2') self._traffic['l3'] = S.getValue(self._tunnel_type.upper() + '_FRAME_L3') self._traffic['l4'] = S.getValue(self._tunnel_type.upper() + '_FRAME_L4') + self._traffic['l2']['dstmac'] = S.getValue('NICS')[1]['mac'] elif len(S.getValue('NICS')) and \ (S.getValue('NICS')[0]['type'] == 'vf' or S.getValue('NICS')[1]['type'] == 'vf'): @@ -181,8 +187,6 @@ class TestCase(object): def run_initialize(self): """ Prepare test execution environment """ - self._logger.debug(self.name) - # mount hugepages if needed self._mount_hugepages() @@ -338,7 +342,7 @@ class TestCase(object): # ...and continue with traffic generation, but keep # in mind, that clean deployment does not configure # OVS nor executes the traffic - if self.deployment != 'clean': + if self.deployment != 'clean' and not self._step_send_traffic: self._traffic_ctl.send_traffic(self._traffic) # dump vswitch flows before they are affected by VNF termination @@ -360,23 +364,6 @@ class TestCase(object): # report test results self.run_report() - # restore original settings - for key in self._settings_original: - S.setValue(key, self._settings_original[key]) - - def _update_settings(self, param, value): - """ Check value of given configuration parameter - In case that new value is different, then testcase - specific settings is updated and original value stored - - :param param: Name of parameter inside settings - :param value: Disired parameter value - """ - orig_value = S.getValue(param) - if orig_value != value: - self._settings_original[param] = copy.deepcopy(orig_value) - S.setValue(param, value) - def _append_results(self, results): """ Method appends mandatory Test Case results to list of dictionaries. @@ -684,18 +671,22 @@ class TestCase(object): if self._step_vnf_list[vnf]: self._step_vnf_list[vnf].stop() - def step_eval_param(self, param, STEP): - # pylint: disable=invalid-name + def step_eval_param(self, param, step_result): """ Helper function for #STEP macro evaluation """ if isinstance(param, str): # evaluate every #STEP reference inside parameter itself - macros = re.findall(r'#STEP\[[\w\[\]\-\'\"]+\]', param) + macros = re.findall(r'(#STEP\[([\w\-:]+)\]((\[[\w\-\'\"]+\])*))', param) + if macros: for macro in macros: + if macro[1] in self._step_result_mapping: + key = self._step_result_mapping[macro[1]] + else: + key = macro[1] # pylint: disable=eval-used - tmp_val = str(eval(macro[1:])) - param = param.replace(macro, tmp_val) + tmp_val = str(eval('step_result[{}]{}'.format(key, macro[2]))) + param = param.replace(macro[0], tmp_val) # evaluate references to vsperf configuration options macros = re.findall(r'\$(([\w\-]+)(\[[\w\[\]\-\'\"]+\])*)', param) @@ -713,12 +704,12 @@ class TestCase(object): elif isinstance(param, list) or isinstance(param, tuple): tmp_list = [] for item in param: - tmp_list.append(self.step_eval_param(item, STEP)) + tmp_list.append(self.step_eval_param(item, step_result)) return tmp_list elif isinstance(param, dict): tmp_dict = {} for (key, value) in param.items(): - tmp_dict[key] = self.step_eval_param(value, STEP) + tmp_dict[key] = self.step_eval_param(value, step_result) return tmp_dict else: return param @@ -732,6 +723,7 @@ class TestCase(object): eval_params.append(self.step_eval_param(param, step_result)) return eval_params + # pylint: disable=too-many-locals, too-many-branches, too-many-statements def step_run(self): """ Execute actions specified by TestSteps list @@ -753,6 +745,30 @@ class TestCase(object): # run test step by step... for i, step in enumerate(self.test): step_ok = not self._step_check + step_check = self._step_check + regex = None + # configure step result mapping if step alias/label is detected + if step[0].startswith('#'): + key = step[0][1:] + if key.isdigit(): + raise RuntimeError('Step alias can\'t be an integer value {}'.format(key)) + if key in self._step_result_mapping: + raise RuntimeError('Step alias {} has been used already for step ' + '{}'.format(key, self._step_result_mapping[key])) + self._step_result_mapping[step[0][1:]] = i + step = step[1:] + + # store regex filter if it is specified + if isinstance(step[-1], str) and step[-1].startswith('|'): + # evalute macros and variables used in regex + regex = self.step_eval_params([step[-1][1:]], self._step_result[:i])[0] + step = step[:-1] + + # check if step verification should be suppressed + if step[0].startswith('!'): + step_check = False + step_ok = True + step[0] = step[0][1:] if step[0] == 'vswitch': test_object = self._vswitch_ctl.get_vswitch() elif step[0] == 'namespace': @@ -772,6 +788,9 @@ class TestCase(object): tmp_traffic = copy.deepcopy(self._traffic) tmp_traffic.update(step[2]) step[2] = tmp_traffic + # store indication that traffic has been sent + # so it is not sent again after the execution of teststeps + self._step_send_traffic = True elif step[0].startswith('vnf'): if not self._step_vnf_list[step[0]]: # initialize new VM @@ -785,6 +804,15 @@ class TestCase(object): self._logger.debug("Sleep %s seconds", step[1]) time.sleep(int(step[1])) continue + elif step[0] == 'log': + test_object = self._logger + # there isn't a need for validation of log entry + step_check = False + step_ok = True + elif step[0] == 'pdb': + import pdb + pdb.set_trace() + continue else: self._logger.error("Unsupported test object %s", step[0]) self._step_status = {'status' : False, 'details' : ' '.join(step)} @@ -793,7 +821,7 @@ class TestCase(object): return False test_method = getattr(test_object, step[1]) - if self._step_check: + if step_check: test_method_check = getattr(test_object, CHECK_PREFIX + step[1]) else: test_method_check = None @@ -804,18 +832,24 @@ class TestCase(object): # to support negative indexes step_params = self.step_eval_params(step[2:], self._step_result[:i]) step_log = '{} {}'.format(' '.join(step[:2]), step_params) + step_log += ' filter "{}"'.format(regex) if regex else '' self._logger.debug("Step %s '%s' start", i, step_log) self._step_result[i] = test_method(*step_params) + if regex: + # apply regex to step output + self._step_result[i] = functions.filter_output( + self._step_result[i], regex) + self._logger.debug("Step %s '%s' results '%s'", i, step_log, self._step_result[i]) time.sleep(S.getValue('TEST_STEP_DELAY')) - if self._step_check: + if step_check: step_ok = test_method_check(self._step_result[i], *step_params) except (AssertionError, AttributeError, IndexError) as ex: step_ok = False self._logger.error("Step %s raised %s", i, type(ex).__name__) - if self._step_check: + if step_check: self.step_report_status("Step {} - '{}'".format(i, step_log), step_ok) if not step_ok: |