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 time
  18 import csv
  19 import os
  20 import logging
  21 from collections import OrderedDict
  22
  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         No external actions yet.
  35         """
  36         self._logger = logging.getLogger(__name__)
  37         self.name = cfg['Name']
  38         self.desc = cfg.get('Description', 'No description given.')
  39         self._traffic_type = cfg['Traffic Type']
  40         self._deployment = cfg['Deployment']
  41         self._collector = cfg['Collector']
  42         self._results_dir = results_dir
  43
  44     def run(self):
  45         """Run the test
  46
  47         All setup and teardown through controllers is included.
  48         """
  49         self._logger.debug(self.name)
  50
  51         self._logger.debug("Controllers:")
  52         loader = Loader()
  53         traffic_ctl = component_factory.create_traffic(
  54             self._traffic_type,
  55             loader.get_trafficgen_class())
  56         vnf_ctl = component_factory.create_vnf(
  57             self._deployment,
  58             loader.get_vnf_class())
  59         vswitch_ctl = component_factory.create_vswitch(
  60             self._deployment,
  61             loader.get_vswitch_class())
  62         collector_ctl = component_factory.create_collector(
  63             self._collector,
  64             loader.get_collector_class())
  65
  66         self._logger.debug("Setup:")
  67         collector_ctl.log_cpu_stats()
  68         with vswitch_ctl:
  69             if vnf_ctl:
  70                 vnf_ctl.start()
  71                 traffic = {'traffic_type': self._traffic_type}
  72             with traffic_ctl:
  73                 traffic_ctl.send_traffic(traffic)
  74
  75
  76         self._logger.debug("Traffic Results:")
  77         traffic_ctl.print_results()
  78
  79         self._logger.debug("Collector Results:")
  80         self._logger.debug(collector_ctl.get_results())
  81
  82
  83         output_file = "result_" + self.name + "_" + self._deployment +".csv"
  84
  85         self._write_result_to_file(
  86             traffic_ctl.get_results(),
  87             os.path.join(self._results_dir, output_file))
  88
  89     @staticmethod
  90     def _write_result_to_file(results, output):
  91         """Write list of dictionaries to a CSV file.
  92
  93         Each element on list will create separate row in output file.
  94         If output file already exists, data will be appended at the end,
  95         otherwise it will be created.
  96
  97         :param results: list of dictionaries.
  98         :param output: path to output file.
  99         """
 100         with open(output, 'a') as csvfile:
 101
 102             logging.info("Write results to file: " + output)
 103             fieldnames = TestCase._get_unique_keys(results)
 104
 105             writer = csv.DictWriter(csvfile, fieldnames)
 106
 107             if not csvfile.tell():  # file is now empty
 108                 writer.writeheader()
 109
 110             for result in results:
 111                 writer.writerow(result)
 112
 113
 114     @staticmethod
 115     def _get_unique_keys(list_of_dicts):
 116         """Gets unique key values as ordered list of strings in given dicts
 117
 118         :param list_of_dicts: list of dictionaries.
 119
 120         :returns: list of unique keys(strings).
 121         """
 122         result = OrderedDict()
 123         for item in list_of_dicts:
 124             for key in item.keys():
 125                 result[key] = ''
 126
 127         return list(result.keys())