DOWNLINK = 'downlink'
DROP_PERCENT_ROUND = 6
RATE_ROUND = 5
+ STATUS_SUCCESS = "Success"
+ STATUS_FAIL = "Failure"
def __init__(self, yaml_data):
super(IXIARFC2544Profile, self).__init__(yaml_data)
self._ixia_traffic_generate(traffic, ixia_obj)
return first_run
- def get_drop_percentage(self, samples, tol_min, tolerance,
+ def get_drop_percentage(self, samples, tol_min, tolerance, precision,
first_run=False):
completed = False
drop_percent = 100
else:
completed = True
+ LOG.debug("tolerance=%s, tolerance_precision=%s drop_percent=%s "
+ "completed=%s", tolerance, precision, drop_percent,
+ completed)
+
latency_ns_avg = float(
sum([samples[iface]['Store-Forward_Avg_latency_ns']
for iface in samples])) / num_ifaces
sum([samples[iface]['Store-Forward_Max_latency_ns']
for iface in samples])) / num_ifaces
+ samples['Status'] = self.STATUS_FAIL
+ if round(drop_percent, precision) <= tolerance:
+ samples['Status'] = self.STATUS_SUCCESS
+
samples['TxThroughput'] = tx_throughput
samples['RxThroughput'] = rx_throughput
samples['DropPercentage'] = drop_percent
# limitations under the License.
import logging
+import decimal
from multiprocessing import Queue, Value, Process
import os
import posixpath
self._rfc2544 = None
self._tolerance_low = None
self._tolerance_high = None
+ self._tolerance_precision = None
@property
def rfc2544(self):
self.get_rfc_tolerance()
return self._tolerance_high
+ @property
+ def tolerance_precision(self):
+ if self._tolerance_precision is None:
+ self.get_rfc_tolerance()
+ return self._tolerance_precision
+
@property
def correlated_traffic(self):
if self._correlated_traffic is None:
def get_rfc_tolerance(self):
tolerance_str = self.get_rfc2544('allowed_drop_rate', self.DEFAULT_TOLERANCE)
- tolerance_iter = iter(sorted(float(t.strip()) for t in tolerance_str.split('-')))
- self._tolerance_low = next(tolerance_iter)
- self._tolerance_high = next(tolerance_iter, self.tolerance_low)
+ tolerance_iter = iter(sorted(
+ decimal.Decimal(t.strip()) for t in tolerance_str.split('-')))
+ tolerance_low = next(tolerance_iter)
+ tolerance_high = next(tolerance_iter, tolerance_low)
+ self._tolerance_precision = abs(tolerance_high.as_tuple().exponent)
+ self._tolerance_high = float(tolerance_high)
+ self._tolerance_low = float(tolerance_low)
class SampleVNFDeployHelper(object):
min_tol = self.rfc_helper.tolerance_low
max_tol = self.rfc_helper.tolerance_high
+ precision = self.rfc_helper.tolerance_precision
default = "00:00:00:00:00:00"
self._build_ports()
traffic_profile.config.duration)
completed, samples = traffic_profile.get_drop_percentage(
- samples, min_tol, max_tol, first_run=first_run)
+ samples, min_tol, max_tol, precision, first_run=first_run)
self._queue.put(samples)
if completed:
'Store-Forward_Max_latency_ns': 28}
}
rfc2544_profile = ixia_rfc2544.IXIARFC2544Profile(self.TRAFFIC_PROFILE)
- completed, samples = rfc2544_profile.get_drop_percentage(samples, 0, 1)
+ completed, samples = rfc2544_profile.get_drop_percentage(
+ samples, 0, 1, 4)
self.assertTrue(completed)
self.assertEqual(66.9, samples['TxThroughput'])
self.assertEqual(66.833, samples['RxThroughput'])
rfc2544_profile = ixia_rfc2544.IXIARFC2544Profile(self.TRAFFIC_PROFILE)
rfc2544_profile.rate = 1000
completed, samples = rfc2544_profile.get_drop_percentage(
- samples, 0, 0.05)
+ samples, 0, 0.05, 4)
self.assertFalse(completed)
self.assertEqual(66.9, samples['TxThroughput'])
self.assertEqual(66.833, samples['RxThroughput'])
rfc2544_profile = ixia_rfc2544.IXIARFC2544Profile(self.TRAFFIC_PROFILE)
rfc2544_profile.rate = 1000
completed, samples = rfc2544_profile.get_drop_percentage(
- samples, 0.2, 1)
+ samples, 0.2, 1, 4)
self.assertFalse(completed)
self.assertEqual(66.9, samples['TxThroughput'])
self.assertEqual(66.833, samples['RxThroughput'])
rfc2544_profile = ixia_rfc2544.IXIARFC2544Profile(self.TRAFFIC_PROFILE)
rfc2544_profile.rate = 1000
completed, samples = rfc2544_profile.get_drop_percentage(
- samples, 0.2, 1)
+ samples, 0.2, 1, 4)
self.assertFalse(completed)
self.assertEqual(0.0, samples['TxThroughput'])
self.assertEqual(66.833, samples['RxThroughput'])
}
rfc2544_profile = ixia_rfc2544.IXIARFC2544Profile(self.TRAFFIC_PROFILE)
completed, samples = rfc2544_profile.get_drop_percentage(
- samples, 0, 1, first_run=True)
+ samples, 0, 1, 4, first_run=True)
self.assertTrue(completed)
self.assertEqual(66.9, samples['TxThroughput'])
self.assertEqual(66.833, samples['RxThroughput'])
self.assertIsNone(rfc2544_resource_helper._tolerance_high)
self.assertEqual(rfc2544_resource_helper.tolerance_high, 0.15)
self.assertEqual(rfc2544_resource_helper._tolerance_high, 0.15)
+ self.assertEqual(rfc2544_resource_helper._tolerance_precision, 2)
scenario_helper.scenario_cfg = {} # ensure that resource_helper caches
self.assertEqual(rfc2544_resource_helper.tolerance_high, 0.15)
rfc2544_resource_helper = Rfc2544ResourceHelper(scenario_helper)
self.assertEqual(rfc2544_resource_helper.tolerance_high, 0.2)
+ self.assertEqual(rfc2544_resource_helper._tolerance_precision, 1)
def test_property_tolerance_low_not_range(self):
scenario_helper = ScenarioHelper('name1')