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

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