1 # Copyright 2015 Intel Corporation.
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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
21 from collections import OrderedDict
23 import core.component_factory as component_factory
24 from core.loader import Loader
26 class TestCase(object):
27 """TestCase base class
29 In this basic form runs RFC2544 throughput test
31 def __init__(self, cfg, results_dir):
32 """Pull out fields from test config
34 No external actions yet.
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._bidir = cfg['biDirectional']
43 self._results_dir = results_dir
48 All setup and teardown through controllers is included.
50 self._logger.debug(self.name)
52 self._logger.debug("Controllers:")
54 traffic_ctl = component_factory.create_traffic(
56 loader.get_trafficgen_class())
57 vnf_ctl = component_factory.create_vnf(
59 loader.get_vnf_class())
60 vswitch_ctl = component_factory.create_vswitch(
62 loader.get_vswitch_class())
63 collector_ctl = component_factory.create_collector(
65 loader.get_collector_class())
67 self._logger.debug("Setup:")
68 collector_ctl.log_cpu_stats()
72 traffic = {'traffic_type': self._traffic_type, 'bidir': self._bidir}
74 traffic_ctl.send_traffic(traffic)
77 self._logger.debug("Traffic Results:")
78 traffic_ctl.print_results()
80 self._logger.debug("Collector Results:")
81 self._logger.debug(collector_ctl.get_results())
84 output_file = "result_" + self.name + "_" + self._deployment +".csv"
86 self._write_result_to_file(
87 traffic_ctl.get_results(),
88 os.path.join(self._results_dir, output_file))
91 def _write_result_to_file(results, output):
92 """Write list of dictionaries to a CSV file.
94 Each element on list will create separate row in output file.
95 If output file already exists, data will be appended at the end,
96 otherwise it will be created.
98 :param results: list of dictionaries.
99 :param output: path to output file.
101 with open(output, 'a') as csvfile:
103 logging.info("Write results to file: " + output)
104 fieldnames = TestCase._get_unique_keys(results)
106 writer = csv.DictWriter(csvfile, fieldnames)
108 if not csvfile.tell(): # file is now empty
111 for result in results:
112 writer.writerow(result)
116 def _get_unique_keys(list_of_dicts):
117 """Gets unique key values as ordered list of strings in given dicts
119 :param list_of_dicts: list of dictionaries.
121 :returns: list of unique keys(strings).
123 result = OrderedDict()
124 for item in list_of_dicts:
125 for key in item.keys():
128 return list(result.keys())