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 | 210 |
1 files changed, 201 insertions, 9 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..7247ee8be 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') @@ -430,7 +432,6 @@ class TestIXIATrafficGen(unittest.TestCase): class TestIxiaBasicScenario(unittest.TestCase): - def setUp(self): self._mock_IxNextgen = mock.patch.object(ixnet_api, 'IxNextgen') self.mock_IxNextgen = self._mock_IxNextgen.start() @@ -448,15 +449,15 @@ class TestIxiaBasicScenario(unittest.TestCase): self.assertIsInstance(self.scenario, tg_rfc2544_ixia.IxiaBasicScenario) self.assertEqual(self.scenario.client, self.mock_IxNextgen) - def test_apply_config(self): - self.assertIsNone(self.scenario.apply_config()) - def test_create_traffic_model(self): self.mock_IxNextgen.get_vports.return_value = [1, 2, 3, 4] self.scenario.create_traffic_model() self.scenario.client.get_vports.assert_called_once() self.scenario.client.create_traffic_model.assert_called_once_with([1, 3], [2, 4]) + def test_apply_config(self): + self.assertIsNone(self.scenario.apply_config()) + def test_run_protocols(self): self.assertIsNone(self.scenario.run_protocols()) @@ -464,6 +465,97 @@ class TestIxiaBasicScenario(unittest.TestCase): self.assertIsNone(self.scenario.stop_protocols()) +class TestIxiaL3Scenario(TestIxiaBasicScenario): + IXIA_CFG = { + 'flow': { + 'src_ip': ['192.168.0.1-192.168.0.50'], + 'dst_ip': ['192.168.1.1-192.168.1.150'] + } + } + + CONTEXT_CFG = { + 'nodes': { + 'tg__0': { + 'role': 'IxNet', + 'interfaces': { + 'xe0': { + 'vld_id': 'uplink_0', + 'local_ip': '10.1.1.1', + 'local_mac': 'aa:bb:cc:dd:ee:ff', + 'ifname': 'xe0' + }, + 'xe1': { + 'vld_id': 'downlink_0', + 'local_ip': '20.2.2.2', + 'local_mac': 'bb:bb:cc:dd:ee:ee', + 'ifname': 'xe1' + } + }, + 'routing_table': [{ + 'network': "152.16.100.20", + 'netmask': '255.255.0.0', + 'gateway': '152.16.100.21', + 'if': 'xe0' + }] + } + } + } + + def setUp(self): + super(TestIxiaL3Scenario, self).setUp() + self.ixia_cfg = self.IXIA_CFG + self.context_cfg = self.CONTEXT_CFG + self.scenario = tg_rfc2544_ixia.IxiaL3Scenario(self.mock_IxNextgen, + self.context_cfg, + self.ixia_cfg) + + def test___init___(self): + self.assertIsInstance(self.scenario, tg_rfc2544_ixia.IxiaL3Scenario) + self.assertEqual(self.scenario.client, self.mock_IxNextgen) + + def test_create_traffic_model(self): + self.mock_IxNextgen.get_vports.return_value = ['1', '2'] + self.scenario.create_traffic_model() + self.scenario.client.get_vports.assert_called_once() + self.scenario.client.create_ipv4_traffic_model.\ + assert_called_once_with(['1/protocols/static'], + ['2/protocols/static']) + + def test_apply_config(self): + self.scenario._add_interfaces = mock.Mock() + self.scenario._add_static_ips = mock.Mock() + self.assertIsNone(self.scenario.apply_config()) + + def test__add_static(self): + self.mock_IxNextgen.get_vports.return_value = ['1', '2'] + self.mock_IxNextgen.get_static_interface.side_effect = ['intf1', + 'intf2'] + + self.scenario._add_static_ips() + + self.mock_IxNextgen.get_static_interface.assert_any_call('1') + self.mock_IxNextgen.get_static_interface.assert_any_call('2') + + self.scenario.client.add_static_ipv4.assert_any_call( + 'intf1', '1', '192.168.0.1', 49) + self.scenario.client.add_static_ipv4.assert_any_call( + 'intf2', '2', '192.168.1.1', 149) + + def test__add_interfaces(self): + self.mock_IxNextgen.get_vports.return_value = ['1', '2'] + + self.scenario._add_interfaces() + + self.mock_IxNextgen.add_interface.assert_any_call('1', + '10.1.1.1', + 'aa:bb:cc:dd:ee:ff', + '152.16.100.21') + self.mock_IxNextgen.add_interface.assert_any_call('2', + '20.2.2.2', + 'bb:bb:cc:dd:ee:ee', + None) + + class TestIxiaPppoeClientScenario(unittest.TestCase): IXIA_CFG = { @@ -524,12 +616,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 |