yardstick/tests/unit/network_services/traffic_profile/test_http_ixload.py

   1 # Copyright (c) 2017-2019 Intel Corporation
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #      http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14
  15 import unittest
  16 import mock
  17
  18 from oslo_serialization import jsonutils
  19
  20 from yardstick.network_services.traffic_profile import http_ixload
  21 from yardstick.network_services.traffic_profile.http_ixload import \
  22     join_non_strings, validate_non_string_sequence
  23
  24
  25 class TestJoinNonStrings(unittest.TestCase):
  26
  27     def test_validate_non_string_sequence(self):
  28         self.assertEqual(validate_non_string_sequence([1, 2, 3]), [1, 2, 3])
  29         self.assertIsNone(validate_non_string_sequence('123'))
  30         self.assertIsNone(validate_non_string_sequence(1))
  31
  32         self.assertEqual(validate_non_string_sequence(1, 2), 2)
  33         self.assertEqual(validate_non_string_sequence(1, default=2), 2)
  34
  35         with self.assertRaises(RuntimeError):
  36             validate_non_string_sequence(1, raise_exc=RuntimeError)
  37
  38     def test_join_non_strings(self):
  39         self.assertEqual(join_non_strings(':'), '')
  40         self.assertEqual(join_non_strings(':', 'a'), 'a')
  41         self.assertEqual(join_non_strings(':', 'a', 2, 'c'), 'a:2:c')
  42         self.assertEqual(join_non_strings(':', ['a', 2, 'c']), 'a:2:c')
  43         self.assertEqual(join_non_strings(':', 'abc'), 'abc')
  44
  45
  46 class TestIxLoadTrafficGen(unittest.TestCase):
  47
  48     def setUp(self):
  49         ports = [1, 2, 3]
  50         self.test_input = {
  51             "remote_server": "REMOTE_SERVER",
  52             "ixload_cfg": "IXLOAD_CFG",
  53             "result_dir": "RESULT_DIR",
  54             "ixia_chassis": "IXIA_CHASSIS",
  55             "IXIA": {
  56                 "card": "CARD",
  57                 "ports": ports,
  58             },
  59             'links_param': {
  60                 "uplink_0": {
  61                     "ip": {"address": "address",
  62                            "gateway": "gateway",
  63                            "subnet_prefix": "subnet_prefix",
  64                            "mac": "mac"
  65                            }}}
  66         }
  67
  68     def test_parse_run_test(self):
  69         ports = [1, 2, 3]
  70         test_input = {
  71             "remote_server": "REMOTE_SERVER",
  72             "ixload_cfg": "IXLOAD_CFG",
  73             "result_dir": "RESULT_DIR",
  74             "ixia_chassis": "IXIA_CHASSIS",
  75             "IXIA": {
  76                 "card": "CARD",
  77                 "ports": ports,
  78             },
  79             'links_param': {}
  80         }
  81         j = jsonutils.dump_as_bytes(test_input)
  82         ixload = http_ixload.IXLOADHttpTest(j)
  83         self.assertDictEqual(ixload.test_input, test_input)
  84         self.assertIsNone(ixload.parse_run_test())
  85         self.assertEqual(ixload.ports_to_reassign, [
  86             ["IXIA_CHASSIS", "CARD", 1],
  87             ["IXIA_CHASSIS", "CARD", 2],
  88             ["IXIA_CHASSIS", "CARD", 3],
  89         ])
  90         self.assertEqual({}, ixload.links_param)
  91
  92     def test_format_ports_for_reassignment(self):
  93         ports = [
  94             ["IXIA_CHASSIS", "CARD", 1],
  95             ["IXIA_CHASSIS", "CARD", 2],
  96             ["IXIA_CHASSIS", "CARD", 3],
  97         ]
  98         formatted = http_ixload.IXLOADHttpTest.format_ports_for_reassignment(ports)
  99         self.assertEqual(formatted, [
 100             "IXIA_CHASSIS;CARD;1",
 101             "IXIA_CHASSIS;CARD;2",
 102             "IXIA_CHASSIS;CARD;3",
 103         ])
 104
 105     def test_reassign_ports(self):
 106         ports = [1, 2, 3]
 107         test_input = {
 108             "remote_server": "REMOTE_SERVER",
 109             "ixload_cfg": "IXLOAD_CFG",
 110             "result_dir": "RESULT_DIR",
 111             "ixia_chassis": "IXIA_CHASSIS",
 112             "IXIA": {
 113                 "card": "CARD",
 114                 "ports": ports,
 115             },
 116             'links_param': {}
 117         }
 118         j = jsonutils.dump_as_bytes(test_input)
 119         ixload = http_ixload.IXLOADHttpTest(j)
 120         repository = mock.Mock()
 121         test = mock.MagicMock()
 122         test.setPorts = mock.Mock()
 123         ports_to_reassign = [(1, 2, 3), (1, 2, 4)]
 124         ixload.format_ports_for_reassignment = mock.Mock(return_value=["1;2;3"])
 125         self.assertIsNone(ixload.reassign_ports(test, repository, ports_to_reassign))
 126
 127     def test_reassign_ports_error(self):
 128         ports = [1, 2, 3]
 129         test_input = {
 130             "remote_server": "REMOTE_SERVER",
 131             "ixload_cfg": "IXLOAD_CFG",
 132             "result_dir": "RESULT_DIR",
 133             "ixia_chassis": "IXIA_CHASSIS",
 134             "IXIA": {
 135                 "card": "CARD",
 136                 "ports": ports,
 137             },
 138             'links_param': {}
 139         }
 140         j = jsonutils.dump_as_bytes(test_input)
 141         ixload = http_ixload.IXLOADHttpTest(j)
 142         repository = mock.Mock()
 143         test = "test"
 144         ports_to_reassign = [(1, 2, 3), (1, 2, 4)]
 145         ixload.format_ports_for_reassignment = mock.Mock(return_value=["1;2;3"])
 146         ixload.ix_load = mock.MagicMock()
 147         ixload.ix_load.delete = mock.Mock()
 148         ixload.ix_load.disconnect = mock.Mock()
 149         with self.assertRaises(Exception):
 150             ixload.reassign_ports(test, repository, ports_to_reassign)
 151
 152     def test_stat_collector(self):
 153         args = [0, 1]
 154         self.assertIsNone(
 155             http_ixload.IXLOADHttpTest.stat_collector(*args))
 156
 157     def test_IxL_StatCollectorCommand(self):
 158         args = [[0, 1, 2, 3], [0, 1, 2, 3]]
 159         self.assertIsNone(
 160             http_ixload.IXLOADHttpTest.IxL_StatCollectorCommand(*args))
 161
 162     def test_set_results_dir(self):
 163         test_stat_collector = mock.MagicMock()
 164         test_stat_collector.setResultDir = mock.Mock()
 165         results_on_windows = "c:/Results"
 166         self.assertIsNone(
 167             http_ixload.IXLOADHttpTest.set_results_dir(test_stat_collector,
 168                                                        results_on_windows))
 169
 170     def test_set_results_dir_error(self):
 171         test_stat_collector = ""
 172         results_on_windows = "c:/Results"
 173         with self.assertRaises(Exception):
 174             http_ixload.IXLOADHttpTest.set_results_dir(test_stat_collector, results_on_windows)
 175
 176     def test_load_config_file(self):
 177         ports = [1, 2, 3]
 178         test_input = {
 179             "remote_server": "REMOTE_SERVER",
 180             "ixload_cfg": "IXLOAD_CFG",
 181             "result_dir": "RESULT_DIR",
 182             "ixia_chassis": "IXIA_CHASSIS",
 183             "IXIA": {
 184                 "card": "CARD",
 185                 "ports": ports,
 186             },
 187             'links_param': {}
 188         }
 189         j = jsonutils.dump_as_bytes(test_input)
 190         ixload = http_ixload.IXLOADHttpTest(j)
 191         ixload.ix_load = mock.MagicMock()
 192         ixload.ix_load.new = mock.Mock(return_value="")
 193         self.assertIsNotNone(ixload.load_config_file("ixload.cfg"))
 194
 195     def test_load_config_file_error(self):
 196         ports = [1, 2, 3]
 197         test_input = {
 198             "remote_server": "REMOTE_SERVER",
 199             "ixload_cfg": "IXLOAD_CFG",
 200             "result_dir": "RESULT_DIR",
 201             "ixia_chassis": "IXIA_CHASSIS",
 202             "IXIA": {
 203                 "card": "CARD",
 204                 "ports": ports,
 205             },
 206             'links_param': {}
 207         }
 208         j = jsonutils.dump_as_bytes(test_input)
 209         ixload = http_ixload.IXLOADHttpTest(j)
 210         ixload.ix_load = "test"
 211         with self.assertRaises(Exception):
 212             ixload.load_config_file("ixload.cfg")
 213
 214     @mock.patch('yardstick.network_services.traffic_profile.http_ixload.StatCollectorUtils')
 215     @mock.patch('yardstick.network_services.traffic_profile.http_ixload.IxLoad')
 216     def test_start_http_test_connect_error(self, mock_ixload_type, *args):
 217         ports = [1, 2, 3]
 218         test_input = {
 219             "remote_server": "REMOTE_SERVER",
 220             "ixload_cfg": "IXLOAD_CFG",
 221             "result_dir": "RESULT_DIR",
 222             "ixia_chassis": "IXIA_CHASSIS",
 223             "IXIA": {
 224                 "card": "CARD",
 225                 "ports": ports,
 226             },
 227             'links_param': {}
 228         }
 229
 230         j = jsonutils.dump_as_bytes(test_input)
 231
 232         mock_ixload_type.return_value.connect.side_effect = RuntimeError
 233
 234         ixload = http_ixload.IXLOADHttpTest(j)
 235         ixload.results_on_windows = 'windows_result_dir'
 236         ixload.result_dir = 'my_result_dir'
 237
 238         with self.assertRaises(RuntimeError):
 239             ixload.start_http_test()
 240
 241     def test_update_config(self):
 242         net_taraffic_0 = mock.Mock()
 243         net_taraffic_0.name = "HTTP client@uplink_0"
 244         net_taraffic_1 = mock.Mock()
 245         net_taraffic_1.name = "HTTP client@uplink_1"
 246
 247         community_list = [net_taraffic_0, net_taraffic_1, Exception]
 248         ixload = http_ixload.IXLOADHttpTest(
 249             jsonutils.dump_as_bytes(self.test_input))
 250
 251         ixload.links_param = {"uplink_0": {"ip": {},
 252                                            "http_client": {}}}
 253
 254         ixload.test = mock.Mock()
 255         ixload.test.communityList = community_list
 256
 257         ixload.update_network_param = mock.Mock()
 258         ixload.update_http_client_param = mock.Mock()
 259
 260         ixload.update_config()
 261
 262         ixload.update_network_param.assert_called_once_with(net_taraffic_0, {})
 263         ixload.update_http_client_param.assert_called_once_with(net_taraffic_0,
 264                                                                 {})
 265
 266     def test_update_network_mac_address(self):
 267         ethernet = mock.MagicMock()
 268         net_traffic = mock.Mock()
 269         net_traffic.network.getL1Plugin.return_value = ethernet
 270
 271         ixload = http_ixload.IXLOADHttpTest(
 272             jsonutils.dump_as_bytes(self.test_input))
 273
 274         ix_net_l2_ethernet_plugin = ethernet.childrenList[0]
 275         ix_net_ip_v4_v6_plugin = ix_net_l2_ethernet_plugin.childrenList[0]
 276         ix_net_ip_v4_v6_range = ix_net_ip_v4_v6_plugin.rangeList[0]
 277
 278         ixload.update_network_mac_address(net_traffic, "auto")
 279         ix_net_ip_v4_v6_range.config.assert_called_once_with(
 280             autoMacGeneration=True)
 281
 282         ixload.update_network_mac_address(net_traffic, "00:00:00:00:00:01")
 283         ix_net_ip_v4_v6_range.config.assert_called_with(
 284             autoMacGeneration=False)
 285         mac_range = ix_net_ip_v4_v6_range.getLowerRelatedRange("MacRange")
 286         mac_range.config.assert_called_once_with(mac="00:00:00:00:00:01")
 287
 288         net_traffic.network.getL1Plugin.return_value = Exception
 289
 290         with self.assertRaises(http_ixload.InvalidRxfFile):
 291             ixload.update_network_mac_address(net_traffic, "auto")
 292
 293     def test_update_network_address(self):
 294         ethernet = mock.MagicMock()
 295         net_traffic = mock.Mock()
 296         net_traffic.network.getL1Plugin.return_value = ethernet
 297
 298         ixload = http_ixload.IXLOADHttpTest(
 299             jsonutils.dump_as_bytes(self.test_input))
 300
 301         ix_net_l2_ethernet_plugin = ethernet.childrenList[0]
 302         ix_net_ip_v4_v6_plugin = ix_net_l2_ethernet_plugin.childrenList[0]
 303         ix_net_ip_v4_v6_range = ix_net_ip_v4_v6_plugin.rangeList[0]
 304
 305         ixload.update_network_address(net_traffic, "address", "gateway",
 306                                       "prefix")
 307         ix_net_ip_v4_v6_range.config.assert_called_once_with(
 308             prefix="prefix",
 309             ipAddress="address",
 310             gatewayAddress="gateway")
 311
 312         net_traffic.network.getL1Plugin.return_value = Exception
 313
 314         with self.assertRaises(http_ixload.InvalidRxfFile):
 315             ixload.update_network_address(net_traffic, "address", "gateway",
 316                                           "prefix")
 317
 318     def test_update_network_param(self):
 319         net_traffic = mock.Mock()
 320
 321         ixload = http_ixload.IXLOADHttpTest(
 322             jsonutils.dump_as_bytes(self.test_input))
 323
 324         ixload.update_network_address = mock.Mock()
 325         ixload.update_network_mac_address = mock.Mock()
 326
 327         param = {"address": "address",
 328                  "gateway": "gateway",
 329                  "subnet_prefix": "subnet_prefix",
 330                  "mac": "mac"
 331                  }
 332
 333         ixload.update_network_param(net_traffic, param)
 334
 335         ixload.update_network_address.assert_called_once_with(net_traffic,
 336                                                               "address",
 337                                                               "gateway",
 338                                                               "subnet_prefix")
 339
 340         ixload.update_network_mac_address.assert_called_once_with(
 341             net_traffic,
 342             "mac")
 343
 344     def test_update_http_client_param(self):
 345         net_traffic = mock.Mock()
 346
 347         ixload = http_ixload.IXLOADHttpTest(
 348             jsonutils.dump_as_bytes(self.test_input))
 349
 350         ixload.update_page_size = mock.Mock()
 351         ixload.update_user_count = mock.Mock()
 352
 353         param = {"page_object": "page_object",
 354                  "simulated_users": "simulated_users"}
 355
 356         ixload.update_http_client_param(net_traffic, param)
 357
 358         ixload.update_page_size.assert_called_once_with(net_traffic,
 359                                                         "page_object")
 360         ixload.update_user_count.assert_called_once_with(net_traffic,
 361                                                          "simulated_users")
 362
 363     def test_update_page_size(self):
 364         activity = mock.MagicMock()
 365         net_traffic = mock.Mock()
 366
 367         ixload = http_ixload.IXLOADHttpTest(
 368             jsonutils.dump_as_bytes(self.test_input))
 369
 370         net_traffic.activityList = [activity]
 371         ix_http_command = activity.agent.actionList[0]
 372         ixload.update_page_size(net_traffic, "page_object")
 373         ix_http_command.config.assert_called_once_with(
 374             pageObject="page_object")
 375
 376         net_traffic.activityList = []
 377         with self.assertRaises(http_ixload.InvalidRxfFile):
 378             ixload.update_page_size(net_traffic, "page_object")
 379
 380     def test_update_user_count(self):
 381         activity = mock.MagicMock()
 382         net_traffic = mock.Mock()
 383
 384         ixload = http_ixload.IXLOADHttpTest(
 385             jsonutils.dump_as_bytes(self.test_input))
 386
 387         net_traffic.activityList = [activity]
 388         ixload.update_user_count(net_traffic, 123)
 389         activity.config.assert_called_once_with(userObjectiveValue=123)
 390
 391         net_traffic.activityList = []
 392         with self.assertRaises(http_ixload.InvalidRxfFile):
 393             ixload.update_user_count(net_traffic, 123)
 394
 395     @mock.patch('yardstick.network_services.traffic_profile.http_ixload.IxLoad')
 396     @mock.patch('yardstick.network_services.traffic_profile.http_ixload.StatCollectorUtils')
 397     def test_start_http_test(self, *args):
 398         ports = [1, 2, 3]
 399         test_input = {
 400             "remote_server": "REMOTE_SERVER",
 401             "ixload_cfg": "IXLOAD_CFG",
 402             "result_dir": "RESULT_DIR",
 403             "ixia_chassis": "IXIA_CHASSIS",
 404             "IXIA": {
 405                 "card": "CARD",
 406                 "ports": ports,
 407             },
 408             'links_param': {}
 409         }
 410
 411         j = jsonutils.dump_as_bytes(test_input)
 412
 413         ixload = http_ixload.IXLOADHttpTest(j)
 414         ixload.results_on_windows = 'windows_result_dir'
 415         ixload.result_dir = 'my_result_dir'
 416         ixload.load_config_file = mock.MagicMock()
 417
 418         self.assertIsNone(ixload.start_http_test())
 419
 420     @mock.patch('yardstick.network_services.traffic_profile.http_ixload.IxLoad')
 421     @mock.patch('yardstick.network_services.traffic_profile.http_ixload.StatCollectorUtils')
 422     def test_start_http_test_reassign_error(self, *args):
 423         ports = [1, 2, 3]
 424         test_input = {
 425             "remote_server": "REMOTE_SERVER",
 426             "ixload_cfg": "IXLOAD_CFG",
 427             "result_dir": "RESULT_DIR",
 428             "ixia_chassis": "IXIA_CHASSIS",
 429             "IXIA": {
 430                 "card": "CARD",
 431                 "ports": ports,
 432             },
 433             'links_param': {}
 434         }
 435
 436         j = jsonutils.dump_as_bytes(test_input)
 437
 438         ixload = http_ixload.IXLOADHttpTest(j)
 439         ixload.load_config_file = mock.MagicMock()
 440
 441         reassign_ports = mock.Mock(side_effect=RuntimeError)
 442         ixload.reassign_ports = reassign_ports
 443         ixload.results_on_windows = 'windows_result_dir'
 444         ixload.result_dir = 'my_result_dir'
 445
 446         ixload.start_http_test()
 447         reassign_ports.assert_called_once()
 448
 449     @mock.patch("yardstick.network_services.traffic_profile.http_ixload.IXLOADHttpTest")
 450     def test_main(self, *args):
 451         args = ["1", "2", "3"]
 452         http_ixload.main(args)