-# Copyright (c) 2017 Intel Corporation
+# Copyright (c) 2017-2019 Intel Corporation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
class TestIxLoadTrafficGen(unittest.TestCase):
+ def setUp(self):
+ ports = [1, 2, 3]
+ self.test_input = {
+ "remote_server": "REMOTE_SERVER",
+ "ixload_cfg": "IXLOAD_CFG",
+ "result_dir": "RESULT_DIR",
+ "ixia_chassis": "IXIA_CHASSIS",
+ "IXIA": {
+ "card": "CARD",
+ "ports": ports,
+ },
+ 'links_param': {
+ "uplink_0": {
+ "ip": {"address": "address",
+ "gateway": "gateway",
+ "subnet_prefix": "subnet_prefix",
+ "mac": "mac"
+ }}}
+ }
+
def test_parse_run_test(self):
ports = [1, 2, 3]
test_input = {
"card": "CARD",
"ports": ports,
},
+ 'links_param': {}
}
j = jsonutils.dump_as_bytes(test_input)
ixload = http_ixload.IXLOADHttpTest(j)
["IXIA_CHASSIS", "CARD", 2],
["IXIA_CHASSIS", "CARD", 3],
])
+ self.assertEqual({}, ixload.links_param)
def test_format_ports_for_reassignment(self):
ports = [
"card": "CARD",
"ports": ports,
},
+ 'links_param': {}
}
j = jsonutils.dump_as_bytes(test_input)
ixload = http_ixload.IXLOADHttpTest(j)
"card": "CARD",
"ports": ports,
},
+ 'links_param': {}
}
j = jsonutils.dump_as_bytes(test_input)
ixload = http_ixload.IXLOADHttpTest(j)
"card": "CARD",
"ports": ports,
},
+ 'links_param': {}
}
j = jsonutils.dump_as_bytes(test_input)
ixload = http_ixload.IXLOADHttpTest(j)
"card": "CARD",
"ports": ports,
},
+ 'links_param': {}
}
j = jsonutils.dump_as_bytes(test_input)
ixload = http_ixload.IXLOADHttpTest(j)
"card": "CARD",
"ports": ports,
},
+ 'links_param': {}
}
j = jsonutils.dump_as_bytes(test_input)
with self.assertRaises(RuntimeError):
ixload.start_http_test()
+ def test_update_config(self):
+ net_taraffic_0 = mock.Mock()
+ net_taraffic_0.name = "HTTP client@uplink_0"
+ net_taraffic_1 = mock.Mock()
+ net_taraffic_1.name = "HTTP client@uplink_1"
+
+ community_list = [net_taraffic_0, net_taraffic_1, Exception]
+ ixload = http_ixload.IXLOADHttpTest(
+ jsonutils.dump_as_bytes(self.test_input))
+
+ ixload.links_param = {"uplink_0": {"ip": {},
+ "http_client": {}}}
+
+ ixload.test = mock.Mock()
+ ixload.test.communityList = community_list
+
+ ixload.update_network_param = mock.Mock()
+ ixload.update_http_client_param = mock.Mock()
+
+ ixload.update_config()
+
+ ixload.update_network_param.assert_called_once_with(net_taraffic_0, {})
+ ixload.update_http_client_param.assert_called_once_with(net_taraffic_0,
+ {})
+
+ def test_update_network_mac_address(self):
+ ethernet = mock.MagicMock()
+ net_traffic = mock.Mock()
+ net_traffic.network.getL1Plugin.return_value = ethernet
+
+ ixload = http_ixload.IXLOADHttpTest(
+ jsonutils.dump_as_bytes(self.test_input))
+
+ ix_net_l2_ethernet_plugin = ethernet.childrenList[0]
+ ix_net_ip_v4_v6_plugin = ix_net_l2_ethernet_plugin.childrenList[0]
+ ix_net_ip_v4_v6_range = ix_net_ip_v4_v6_plugin.rangeList[0]
+
+ ixload.update_network_mac_address(net_traffic, "auto")
+ ix_net_ip_v4_v6_range.config.assert_called_once_with(
+ autoMacGeneration=True)
+
+ ixload.update_network_mac_address(net_traffic, "00:00:00:00:00:01")
+ ix_net_ip_v4_v6_range.config.assert_called_with(
+ autoMacGeneration=False)
+ mac_range = ix_net_ip_v4_v6_range.getLowerRelatedRange("MacRange")
+ mac_range.config.assert_called_once_with(mac="00:00:00:00:00:01")
+
+ net_traffic.network.getL1Plugin.return_value = Exception
+
+ with self.assertRaises(http_ixload.InvalidRxfFile):
+ ixload.update_network_mac_address(net_traffic, "auto")
+
+ def test_update_network_address(self):
+ ethernet = mock.MagicMock()
+ net_traffic = mock.Mock()
+ net_traffic.network.getL1Plugin.return_value = ethernet
+
+ ixload = http_ixload.IXLOADHttpTest(
+ jsonutils.dump_as_bytes(self.test_input))
+
+ ix_net_l2_ethernet_plugin = ethernet.childrenList[0]
+ ix_net_ip_v4_v6_plugin = ix_net_l2_ethernet_plugin.childrenList[0]
+ ix_net_ip_v4_v6_range = ix_net_ip_v4_v6_plugin.rangeList[0]
+
+ ixload.update_network_address(net_traffic, "address", "gateway",
+ "prefix")
+ ix_net_ip_v4_v6_range.config.assert_called_once_with(
+ prefix="prefix",
+ ipAddress="address",
+ gatewayAddress="gateway")
+
+ net_traffic.network.getL1Plugin.return_value = Exception
+
+ with self.assertRaises(http_ixload.InvalidRxfFile):
+ ixload.update_network_address(net_traffic, "address", "gateway",
+ "prefix")
+
+ def test_update_network_param(self):
+ net_traffic = mock.Mock()
+
+ ixload = http_ixload.IXLOADHttpTest(
+ jsonutils.dump_as_bytes(self.test_input))
+
+ ixload.update_network_address = mock.Mock()
+ ixload.update_network_mac_address = mock.Mock()
+
+ param = {"address": "address",
+ "gateway": "gateway",
+ "subnet_prefix": "subnet_prefix",
+ "mac": "mac"
+ }
+
+ ixload.update_network_param(net_traffic, param)
+
+ ixload.update_network_address.assert_called_once_with(net_traffic,
+ "address",
+ "gateway",
+ "subnet_prefix")
+
+ ixload.update_network_mac_address.assert_called_once_with(
+ net_traffic,
+ "mac")
+
+ def test_update_http_client_param(self):
+ net_traffic = mock.Mock()
+
+ ixload = http_ixload.IXLOADHttpTest(
+ jsonutils.dump_as_bytes(self.test_input))
+
+ ixload.update_page_size = mock.Mock()
+ ixload.update_user_count = mock.Mock()
+
+ param = {"page_object": "page_object",
+ "simulated_users": "simulated_users"}
+
+ ixload.update_http_client_param(net_traffic, param)
+
+ ixload.update_page_size.assert_called_once_with(net_traffic,
+ "page_object")
+ ixload.update_user_count.assert_called_once_with(net_traffic,
+ "simulated_users")
+
+ def test_update_page_size(self):
+ activity = mock.MagicMock()
+ net_traffic = mock.Mock()
+
+ ixload = http_ixload.IXLOADHttpTest(
+ jsonutils.dump_as_bytes(self.test_input))
+
+ net_traffic.activityList = [activity]
+ ix_http_command = activity.agent.actionList[0]
+ ixload.update_page_size(net_traffic, "page_object")
+ ix_http_command.config.assert_called_once_with(
+ pageObject="page_object")
+
+ net_traffic.activityList = []
+ with self.assertRaises(http_ixload.InvalidRxfFile):
+ ixload.update_page_size(net_traffic, "page_object")
+
+ def test_update_user_count(self):
+ activity = mock.MagicMock()
+ net_traffic = mock.Mock()
+
+ ixload = http_ixload.IXLOADHttpTest(
+ jsonutils.dump_as_bytes(self.test_input))
+
+ net_traffic.activityList = [activity]
+ ixload.update_user_count(net_traffic, 123)
+ activity.config.assert_called_once_with(userObjectiveValue=123)
+
+ net_traffic.activityList = []
+ with self.assertRaises(http_ixload.InvalidRxfFile):
+ ixload.update_user_count(net_traffic, 123)
+
@mock.patch('yardstick.network_services.traffic_profile.http_ixload.IxLoad')
@mock.patch('yardstick.network_services.traffic_profile.http_ixload.StatCollectorUtils')
def test_start_http_test(self, *args):
"card": "CARD",
"ports": ports,
},
+ 'links_param': {}
}
j = jsonutils.dump_as_bytes(test_input)
"card": "CARD",
"ports": ports,
},
+ 'links_param': {}
}
j = jsonutils.dump_as_bytes(test_input)
Code Review / yardstick.git / blobdiff