+
+ @mock.patch.object(ixia_rfc2544.IXIARFC2544PppoeScenarioProfile,
+ '_get_flow_groups_params')
+ def test_update_traffic_profile(self, mock_get_flow_groups_params):
+ networks = {
+ 'uplink_0': 'data1',
+ 'downlink_0': 'data2',
+ 'uplink_1': 'data3',
+ 'downlink_1': 'data4'
+ }
+ ports = ['xe0', 'xe1', 'xe2', 'xe3']
+ mock_traffic_gen = mock.Mock()
+ mock_traffic_gen.networks = networks
+ mock_traffic_gen.vnfd_helper.port_num.side_effect = ports
+ self.ixia_tp.update_traffic_profile(mock_traffic_gen)
+ mock_get_flow_groups_params.assert_called_once()
+ self.assertEqual(self.ixia_tp.ports, ports)
+
+ def test__get_prio_flows_drop_percentage(self):
+
+ input_stats = {
+ '0': {
+ 'InPackets': 50,
+ 'OutPackets': 100,
+ 'Store-Forward_Avg_latency_ns': 10,
+ 'Store-Forward_Min_latency_ns': 10,
+ 'Store-Forward_Max_latency_ns': 10}}
+
+ result = self.ixia_tp._get_prio_flows_drop_percentage(input_stats)
+ self.assertIsNotNone(result['0'].get('DropPercentage'))
+ self.assertEqual(result['0'].get('DropPercentage'), 50.0)
+
+ def test__get_prio_flows_drop_percentage_traffic_not_flowing(self):
+ input_stats = {
+ '0': {
+ 'InPackets': 0,
+ 'OutPackets': 0,
+ 'Store-Forward_Avg_latency_ns': 0,
+ 'Store-Forward_Min_latency_ns': 0,
+ 'Store-Forward_Max_latency_ns': 0}}
+
+ result = self.ixia_tp._get_prio_flows_drop_percentage(input_stats)
+ self.assertIsNotNone(result['0'].get('DropPercentage'))
+ self.assertEqual(result['0'].get('DropPercentage'), 100)
+
+ def test__get_summary_pppoe_subs_counters(self):
+ input_stats = {
+ 'xe0': {
+ 'OutPackets': 100,
+ 'SessionsUp': 4,
+ 'SessionsDown': 0,
+ 'SessionsNotStarted': 0,
+ 'SessionsTotal': 4},
+ 'xe1': {
+ 'OutPackets': 100,
+ 'SessionsUp': 4,
+ 'SessionsDown': 0,
+ 'SessionsNotStarted': 0,
+ 'SessionsTotal': 4}
+ }
+
+ expected_stats = {
+ 'SessionsUp': 8,
+ 'SessionsDown': 0,
+ 'SessionsNotStarted': 0,
+ 'SessionsTotal': 8
+ }
+
+ res = self.ixia_tp._get_summary_pppoe_subs_counters(input_stats)
+ self.assertDictEqual(res, expected_stats)
+
+ @mock.patch.object(ixia_rfc2544.IXIARFC2544PppoeScenarioProfile,
+ '_get_prio_flows_drop_percentage')
+ @mock.patch.object(ixia_rfc2544.IXIARFC2544PppoeScenarioProfile,
+ '_get_summary_pppoe_subs_counters')
+ def test_get_drop_percentage(self, mock_get_pppoe_subs,
+ mock_sum_prio_drop_rate):
+ samples = {
+ 'priority_stats': {
+ '0': {
+ 'InPackets': 100,
+ 'OutPackets': 100,
+ 'InBytes': 6400,
+ 'OutBytes': 6400,
+ 'LatencyAvg': 10,
+ 'LatencyMin': 10,
+ 'LatencyMax': 10}},
+ 'xe0': {
+ 'InPackets': 100,
+ 'OutPackets': 100,
+ 'InBytes': 6400,
+ 'OutBytes': 6400,
+ 'LatencyAvg': 10,
+ 'LatencyMin': 10,
+ 'LatencyMax': 10}}
+
+ mock_get_pppoe_subs.return_value = {'SessionsUp': 1}
+ mock_sum_prio_drop_rate.return_value = {'0': {'DropPercentage': 0.0}}
+
+ self.ixia_tp._get_framesize = mock.Mock(return_value='64B')
+ status, res = self.ixia_tp.get_drop_percentage(
+ samples, tol_min=0.0, tolerance=0.0001, precision=0,
+ resolution=0.1, first_run=True)
+ self.assertIsNotNone(res.get('DropPercentage'))
+ self.assertIsNotNone(res.get('Priority'))
+ self.assertIsNotNone(res.get('SessionsUp'))
+ self.assertEqual(res['DropPercentage'], 0.0)
+ self.assertEqual(res['Rate'], 100.0)
+ self.assertEqual(res['PktSize'], '64B')
+ self.assertTrue(status)
+ mock_sum_prio_drop_rate.assert_called_once()
+ mock_get_pppoe_subs.assert_called_once()
+
+ @mock.patch.object(ixia_rfc2544.IXIARFC2544PppoeScenarioProfile,
+ '_get_prio_flows_drop_percentage')
+ @mock.patch.object(ixia_rfc2544.IXIARFC2544PppoeScenarioProfile,
+ '_get_summary_pppoe_subs_counters')
+ def test_get_drop_percentage_failed_status(self, mock_get_pppoe_subs,
+ mock_sum_prio_drop_rate):
+ samples = {
+ 'priority_stats': {
+ '0': {
+ 'InPackets': 90,
+ 'OutPackets': 100,
+ 'InBytes': 5760,
+ 'OutBytes': 6400,
+ 'LatencyAvg': 10,
+ 'LatencyMin': 10,
+ 'LatencyMax': 10}},
+ 'xe0': {
+ 'InPackets': 90,
+ 'OutPackets': 100,
+ 'InBytes': 5760,
+ 'OutBytes': 6400,
+ 'LatencyAvg': 10,
+ 'LatencyMin': 10,
+ 'LatencyMax': 10}}
+
+ mock_get_pppoe_subs.return_value = {'SessionsUp': 1}
+ mock_sum_prio_drop_rate.return_value = {'0': {'DropPercentage': 0.0}}
+
+ status, res = self.ixia_tp.get_drop_percentage(
+ samples, tol_min=0.0, tolerance=0.0001, precision=0,
+ resolution=0.1, first_run=True)
+ self.assertIsNotNone(res.get('DropPercentage'))
+ self.assertIsNotNone(res.get('Priority'))
+ self.assertIsNotNone(res.get('SessionsUp'))
+ self.assertEqual(res['DropPercentage'], 10.0)
+ self.assertFalse(status)
+ mock_sum_prio_drop_rate.assert_called_once()
+ mock_get_pppoe_subs.assert_called_once()
+
+ @mock.patch.object(ixia_rfc2544.IXIARFC2544PppoeScenarioProfile,
+ '_get_prio_flows_drop_percentage')
+ @mock.patch.object(ixia_rfc2544.IXIARFC2544PppoeScenarioProfile,
+ '_get_summary_pppoe_subs_counters')
+ def test_get_drop_percentage_priority_flow_check(self, mock_get_pppoe_subs,
+ mock_sum_prio_drop_rate):
+ samples = {
+ 'priority_stats': {
+ '0': {
+ 'InPackets': 100,
+ 'OutPackets': 100,
+ 'InBytes': 6400,
+ 'OutBytes': 6400,
+ 'LatencyAvg': 10,
+ 'LatencyMin': 10,
+ 'LatencyMax': 10}},
+ 'xe0': {
+ 'InPackets': 90,
+ 'OutPackets': 100,
+ 'InBytes': 5760,
+ 'OutBytes': 6400,
+ 'LatencyAvg': 10,
+ 'LatencyMin': 10,
+ 'LatencyMax': 10
+ }}
+
+ mock_get_pppoe_subs.return_value = {'SessionsUp': 1}
+ mock_sum_prio_drop_rate.return_value = {'0': {'DropPercentage': 0.0}}
+
+ tc_rfc2544_opts = {'priority': '0',
+ 'allowed_drop_rate': '0.0001 - 0.0001'}
+ status, res = self.ixia_tp.get_drop_percentage(
+ samples, tol_min=15.0000, tolerance=15.0001, precision=0,
+ resolution=0.1, first_run=True, tc_rfc2544_opts=tc_rfc2544_opts)
+ self.assertIsNotNone(res.get('DropPercentage'))
+ self.assertIsNotNone(res.get('Priority'))
+ self.assertIsNotNone(res.get('SessionsUp'))
+ self.assertTrue(status)
+ mock_sum_prio_drop_rate.assert_called_once()
+ mock_get_pppoe_subs.assert_called_once()