self._traffic_gen_class = traffic_gen_class()
self._traffic_started = False
self._traffic_started_call_count = 0
- self._packet_sizes = settings.getValue('TRAFFICGEN_PKT_SIZES')
- self._trials = get_test_param('rfc2544_trials', 1)
+ self._trials = int(get_test_param('rfc2544_trials', 1))
+ self._duration = int(get_test_param('duration', 30))
self._results = []
+ # If set, comma separated packet_sizes value from --test_params
+ # on cli takes precedence over value in settings file.
+ self._packet_sizes = None
+ packet_sizes_cli = get_test_param('pkt_sizes')
+ if packet_sizes_cli:
+ self._packet_sizes = [int(x.strip())
+ for x in packet_sizes_cli.split(',')]
+ else:
+ self._packet_sizes = settings.getValue('TRAFFICGEN_PKT_SIZES')
+
def __enter__(self):
"""Call initialisation function.
"""
:param result_dict: Dictionary containing results from trafficgen
:param packet_size: Packet size value.
- :returns: dictionary of results with addictional entries.
+ :returns: dictionary of results with additional entries.
"""
ret_value = result_dict
- #TODO Old TOIT controller had knowledge about scenario beeing
- #executed, should new controller also fill Configuration & ID,
+ # TODO Old TOIT controller had knowledge about scenario beeing
+ # executed, should new controller also fill Configuration & ID,
# or this should be passed to TestCase?
ret_value[ResultsConstants.TYPE] = 'rfc2544'
ret_value[ResultsConstants.PACKET_SIZE] = str(packet_size)
str(self._traffic_gen_class))
for packet_size in self._packet_sizes:
- traffic['l2'] = {'framesize': packet_size}
+ # Merge framesize with the default traffic definition
+ if 'l2' in traffic:
+ traffic['l2'] = dict(traffic['l2'],
+ **{'framesize': packet_size})
+ else:
+ traffic['l2'] = {'framesize': packet_size}
+
if traffic['traffic_type'] == 'back2back':
result = self._traffic_gen_class.send_rfc2544_back2back(
- traffic, trials=int(self._trials),
- duration=int(get_test_param('rfc2544_duration', 20)))
+ traffic, trials=self._trials, duration=self._duration)
+ elif traffic['traffic_type'] == 'continuous':
+ result = self._traffic_gen_class.send_cont_traffic(
+ traffic, duration=self._duration)
else:
result = self._traffic_gen_class.send_rfc2544_throughput(
- traffic, trials=int(self._trials),
- duration=int(get_test_param('rfc2544_duration', 20)))
+ traffic, trials=self._trials, duration=self._duration)
result = TrafficControllerRFC2544._append_results(result,
packet_size)
traffic['l2'] = {'framesize': packet_size}
self._traffic_gen_class.start_rfc2544_throughput(
traffic,
- trials=int(self._trials),
- duration=int(get_test_param('rfc2544_duration', 20)))
+ trials=self._trials,
+ duration=self._duration)
self._traffic_started = True
if len(function['args']) > 0:
function['function'](function['args'])
logging.info(" Key: " + str(key) +
", Value: " + str(value))
-
def get_results(self):
"""IResult interface implementation.
"""
return self._results
+
+ def validate_send_traffic(self, result, traffic):
+ """Verify that send traffic has succeeded
+ """
+ if len(self._results):
+ if 'b2b_frames' in self._results[-1]:
+ return float(self._results[-1]['b2b_frames']) > 0
+ elif 'throughput_rx_fps' in self._results[-1]:
+ return float(self._results[-1]['throughput_rx_fps']) > 0
+ else:
+ return True
+ else:
+ return False