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
122
123
124
125
126
127
128
|
# Copyright 2015 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.
"""TestCase base class
"""
import time
import csv
import os
import logging
from collections import OrderedDict
import core.component_factory as component_factory
from core.loader import Loader
class TestCase(object):
"""TestCase base class
In this basic form runs RFC2544 throughput test
"""
def __init__(self, cfg, results_dir):
"""Pull out fields from test config
No external actions yet.
"""
self._logger = logging.getLogger(__name__)
self.name = cfg['Name']
self.desc = cfg.get('Description', 'No description given.')
self._traffic_type = cfg['Traffic Type']
self._deployment = cfg['Deployment']
self._collector = cfg['Collector']
self._results_dir = results_dir
def run(self):
"""Run the test
All setup and teardown through controllers is included.
"""
self._logger.debug(self.name)
self._logger.debug("Controllers:")
loader = Loader()
traffic_ctl = component_factory.create_traffic(
self._traffic_type,
loader.get_trafficgen_class())
vnf_ctl = component_factory.create_vnf(
self._deployment,
loader.get_vnf_class())
vswitch_ctl = component_factory.create_vswitch(
self._deployment,
loader.get_vswitch_class())
collector_ctl = component_factory.create_collector(
self._collector,
loader.get_collector_class())
self._logger.debug("Setup:")
collector_ctl.log_cpu_stats()
with vswitch_ctl:
if vnf_ctl:
vnf_ctl.start()
#TODO 'traffic' is placeholder for traffic dict
traffic = {'test': 'rfc2544'}
with traffic_ctl:
traffic_ctl.send_traffic(traffic)
self._logger.debug("Traffic Results:")
traffic_ctl.print_results()
self._logger.debug("Collector Results:")
self._logger.debug(collector_ctl.get_results())
output_file = "result_" + self.name + "_" + self._deployment +".csv"
self._write_result_to_file(
traffic_ctl.get_results(),
os.path.join(self._results_dir, output_file))
@staticmethod
def _write_result_to_file(results, output):
"""Write list of dictionaries to a CSV file.
Each element on list will create separate row in output file.
If output file already exists, data will be appended at the end,
otherwise it will be created.
:param results: list of dictionaries.
:param output: path to output file.
"""
with open(output, 'a') as csvfile:
logging.info("Write results to file: " + output)
fieldnames = TestCase._get_unique_keys(results)
writer = csv.DictWriter(csvfile, fieldnames)
if not csvfile.tell(): # file is now empty
writer.writeheader()
for result in results:
writer.writerow(result)
@staticmethod
def _get_unique_keys(list_of_dicts):
"""Gets unique key values as ordered list of strings in given dicts
:param list_of_dicts: list of dictionaries.
:returns: list of unique keys(strings).
"""
result = OrderedDict()
for item in list_of_dicts:
for key in item.keys():
result[key] = ''
return list(result.keys())
|
