diff options
Diffstat (limited to 'yardstick/tests/unit/network_services/vnf_generic/vnf/test_tg_rfc2544_ixia.py')
| -rw-r--r-- | yardstick/tests/unit/network_services/vnf_generic/vnf/test_tg_rfc2544_ixia.py | 112 |
1 files changed, 107 insertions, 5 deletions
diff --git a/yardstick/tests/unit/network_services/vnf_generic/vnf/test_tg_rfc2544_ixia.py b/yardstick/tests/unit/network_services/vnf_generic/vnf/test_tg_rfc2544_ixia.py index e22398847..65bf56f1e 100644 --- a/yardstick/tests/unit/network_services/vnf_generic/vnf/test_tg_rfc2544_ixia.py +++ b/yardstick/tests/unit/network_services/vnf_generic/vnf/test_tg_rfc2544_ixia.py @@ -18,6 +18,7 @@ import mock import six import unittest import ipaddress +from collections import OrderedDict from yardstick.common import utils from yardstick.common import exceptions @@ -105,6 +106,7 @@ class TestIxiaResourceHelper(unittest.TestCase): ixia_rhelper.run_traffic(mock_tprofile) self.assertEqual('fake_samples', ixia_rhelper._queue.get()) + mock_tprofile.update_traffic_profile.assert_called_once() @mock.patch.object(tg_rfc2544_ixia, 'ixnet_api') @@ -524,12 +526,112 @@ class TestIxiaPppoeClientScenario(unittest.TestCase): mock_apply_core_net_cfg.assert_called_once() mock_apply_access_net_cfg.assert_called_once() - def test_create_traffic_model(self): - self.scenario._access_topologies = 'access' - self.scenario._core_topologies = 'core' - self.scenario.create_traffic_model() + @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario, + '_get_endpoints_src_dst_id_pairs') + @mock.patch.object(tg_rfc2544_ixia.IxiaPppoeClientScenario, + '_get_endpoints_src_dst_obj_pairs') + def test_create_traffic_model(self, mock_obj_pairs, mock_id_pairs): + uplink_endpoints = ['group1', 'group2'] + downlink_endpoints = ['group3', 'group3'] + mock_id_pairs.return_value = ['xe0', 'xe1', 'xe0', 'xe1'] + mock_obj_pairs.return_value = ['group1', 'group3', 'group2', 'group3'] + mock_tp = mock.Mock() + mock_tp.full_profile = {'uplink_0': 'data', + 'downlink_0': 'data', + 'uplink_1': 'data', + 'downlink_1': 'data' + } + self.scenario.create_traffic_model(mock_tp) + mock_id_pairs.assert_called_once_with(mock_tp.full_profile) + mock_obj_pairs.assert_called_once_with(['xe0', 'xe1', 'xe0', 'xe1']) self.scenario.client.create_ipv4_traffic_model.assert_called_once_with( - 'access', 'core') + uplink_endpoints, downlink_endpoints) + + def test__get_endpoints_src_dst_id_pairs(self): + full_tp = OrderedDict([ + ('uplink_0', {'ipv4': {'port': 'xe0'}}), + ('downlink_0', {'ipv4': {'port': 'xe1'}}), + ('uplink_1', {'ipv4': {'port': 'xe0'}}), + ('downlink_1', {'ipv4': {'port': 'xe3'}})]) + endpoints_src_dst_pairs = ['xe0', 'xe1', 'xe0', 'xe3'] + res = self.scenario._get_endpoints_src_dst_id_pairs(full_tp) + self.assertEqual(res, endpoints_src_dst_pairs) + + def test__get_endpoints_src_dst_id_pairs_wrong_flows_number(self): + full_tp = OrderedDict([ + ('uplink_0', {'ipv4': {'port': 'xe0'}}), + ('downlink_0', {'ipv4': {'port': 'xe1'}}), + ('uplink_1', {'ipv4': {'port': 'xe0'}})]) + with self.assertRaises(RuntimeError): + self.scenario._get_endpoints_src_dst_id_pairs(full_tp) + + def test__get_endpoints_src_dst_id_pairs_no_port_key(self): + full_tp = OrderedDict([ + ('uplink_0', {'ipv4': {'id': 1}}), + ('downlink_0', {'ipv4': {'id': 2}})]) + self.assertEqual( + self.scenario._get_endpoints_src_dst_id_pairs(full_tp), []) + + def test__get_endpoints_src_dst_obj_pairs_tp_with_port_key(self): + endpoints_id_pairs = ['xe0', 'xe1', + 'xe0', 'xe1', + 'xe0', 'xe3', + 'xe0', 'xe3'] + ixia_cfg = { + 'pppoe_client': { + 'sessions_per_port': 4, + 'sessions_per_svlan': 1 + }, + 'flow': { + 'src_ip': [{'tg__0': 'xe0'}, {'tg__0': 'xe2'}], + 'dst_ip': [{'tg__0': 'xe1'}, {'tg__0': 'xe3'}] + } + } + + expected_result = ['tp1_dg1', 'tp3_dg1', 'tp1_dg2', 'tp3_dg1', + 'tp1_dg3', 'tp4_dg1', 'tp1_dg4', 'tp4_dg1'] + + self.scenario._ixia_cfg = ixia_cfg + self.scenario._access_topologies = ['topology1', 'topology2'] + self.scenario._core_topologies = ['topology3', 'topology4'] + self.mock_IxNextgen.get_topology_device_groups.side_effect = \ + [['tp1_dg1', 'tp1_dg2', 'tp1_dg3', 'tp1_dg4'], + ['tp2_dg1', 'tp2_dg2', 'tp2_dg3', 'tp2_dg4'], + ['tp3_dg1'], + ['tp4_dg1']] + res = self.scenario._get_endpoints_src_dst_obj_pairs( + endpoints_id_pairs) + self.assertEqual(res, expected_result) + + def test__get_endpoints_src_dst_obj_pairs_default_flows_mapping(self): + endpoints_id_pairs = [] + ixia_cfg = { + 'pppoe_client': { + 'sessions_per_port': 4, + 'sessions_per_svlan': 1 + }, + 'flow': { + 'src_ip': [{'tg__0': 'xe0'}, {'tg__0': 'xe2'}], + 'dst_ip': [{'tg__0': 'xe1'}, {'tg__0': 'xe3'}] + } + } + + expected_result = ['tp1_dg1', 'tp3_dg1', 'tp1_dg2', 'tp3_dg1', + 'tp1_dg3', 'tp3_dg1', 'tp1_dg4', 'tp3_dg1', + 'tp2_dg1', 'tp4_dg1', 'tp2_dg2', 'tp4_dg1', + 'tp2_dg3', 'tp4_dg1', 'tp2_dg4', 'tp4_dg1'] + + self.scenario._ixia_cfg = ixia_cfg + self.scenario._access_topologies = ['topology1', 'topology2'] + self.scenario._core_topologies = ['topology3', 'topology4'] + self.mock_IxNextgen.get_topology_device_groups.side_effect = \ + [['tp1_dg1', 'tp1_dg2', 'tp1_dg3', 'tp1_dg4'], + ['tp2_dg1', 'tp2_dg2', 'tp2_dg3', 'tp2_dg4'], + ['tp3_dg1'], + ['tp4_dg1']] + res = self.scenario._get_endpoints_src_dst_obj_pairs( + endpoints_id_pairs) + self.assertEqual(res, expected_result) def test_run_protocols(self): self.scenario.client.is_protocols_running.return_value = True |