testcases/testcase.py

   1 # Copyright 2015 Intel Corporation.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #   http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14 """TestCase base class
  15 """
  16
  17 import csv
  18 import os
  19 import logging
  20 from collections import OrderedDict
  21
  22 from core.results.results_constants import ResultsConstants
  23 import core.component_factory as component_factory
  24 from core.loader import Loader
  25
  26 class TestCase(object):
  27     """TestCase base class
  28
  29     In this basic form runs RFC2544 throughput test
  30     """
  31     def __init__(self, cfg, results_dir):
  32         """Pull out fields from test config
  33
  34         :param cfg: A dictionary of string-value pairs describing the test
  35             configuration. Both the key and values strings use well-known
  36             values.
  37         :param results_dir: Where the csv formatted results are written.
  38         """
  39         self._logger = logging.getLogger(__name__)
  40         self.name = cfg['Name']
  41         self.desc = cfg.get('Description', 'No description given.')
  42         self._traffic_type = cfg['Traffic Type']
  43         self.deployment = cfg['Deployment']
  44         self._collector = cfg['Collector']
  45         self._bidir = cfg['biDirectional']
  46         self._frame_mod = cfg.get('Frame Modification', None)
  47         if self._frame_mod:
  48             self._frame_mod = self._frame_mod.lower()
  49         self._results_dir = results_dir
  50         self._multistream = cfg.get('MultiStream', 0)
  51
  52     def run(self):
  53         """Run the test
  54
  55         All setup and teardown through controllers is included.
  56         """
  57         self._logger.debug(self.name)
  58
  59         self._logger.debug("Controllers:")
  60         loader = Loader()
  61         traffic_ctl = component_factory.create_traffic(
  62             self._traffic_type,
  63             loader.get_trafficgen_class())
  64         vnf_ctl = component_factory.create_vnf(
  65             self.deployment,
  66             loader.get_vnf_class())
  67         vswitch_ctl = component_factory.create_vswitch(
  68             self.deployment,
  69             loader.get_vswitch_class(),
  70             self._bidir)
  71         collector_ctl = component_factory.create_collector(
  72             self._collector,
  73             loader.get_collector_class())
  74
  75
  76         self._logger.debug("Setup:")
  77         collector_ctl.log_cpu_stats()
  78         with vswitch_ctl:
  79             with vnf_ctl:
  80                 traffic = {'traffic_type': self._traffic_type,
  81                            'bidir': self._bidir,
  82                            'multistream': self._multistream}
  83
  84                 vswitch = vswitch_ctl.get_vswitch()
  85                 if self._frame_mod == "vlan":
  86                     flow = {'table':'2', 'priority':'1000', 'metadata':'2',
  87                             'actions': ['push_vlan:0x8100', 'goto_table:3']}
  88                     vswitch.add_flow('br0', flow)
  89                     flow = {'table':'2', 'priority':'1000', 'metadata':'1',
  90                             'actions': ['push_vlan:0x8100', 'goto_table:3']}
  91                     vswitch.add_flow('br0', flow)
  92
  93                 with traffic_ctl:
  94                     traffic_ctl.send_traffic(traffic)
  95
  96         self._logger.debug("Traffic Results:")
  97         traffic_ctl.print_results()
  98
  99         self._logger.debug("Collector Results:")
 100         self._logger.debug(collector_ctl.get_results())
 101
 102         output_file = "result_" + self.name + "_" + self.deployment +".csv"
 103
 104         TestCase._write_result_to_file(
 105             self._append_results(traffic_ctl.get_results()),
 106             os.path.join(self._results_dir, output_file))
 107
 108     def _append_results(self, results):
 109         """
 110         Method appends mandatory Test Case results to list of dictionaries.
 111
 112         :param results: list of dictionaries which contains results from
 113                 traffic generator.
 114
 115         :returns: modified list of dictionaries.
 116         """
 117         for item in results:
 118             item[ResultsConstants.ID] = self.name
 119             item[ResultsConstants.DEPLOYMENT] = self.deployment
 120
 121         return results
 122
 123
 124     @staticmethod
 125     def _write_result_to_file(results, output):
 126         """Write list of dictionaries to a CSV file.
 127
 128         Each element on list will create separate row in output file.
 129         If output file already exists, data will be appended at the end,
 130         otherwise it will be created.
 131
 132         :param results: list of dictionaries.
 133         :param output: path to output file.
 134         """
 135         with open(output, 'a') as csvfile:
 136
 137             logging.info("Write results to file: " + output)
 138             fieldnames = TestCase._get_unique_keys(results)
 139
 140             writer = csv.DictWriter(csvfile, fieldnames)
 141
 142             if not csvfile.tell():  # file is now empty
 143                 writer.writeheader()
 144
 145             for result in results:
 146                 writer.writerow(result)
 147
 148
 149     @staticmethod
 150     def _get_unique_keys(list_of_dicts):
 151         """Gets unique key values as ordered list of strings in given dicts
 152
 153         :param list_of_dicts: list of dictionaries.
 154
 155         :returns: list of unique keys(strings).
 156         """
 157         result = OrderedDict()
 158         for item in list_of_dicts:
 159             for key in item.keys():
 160                 result[key] = ''
 161
 162         return list(result.keys())