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
  16 from __future__ import absolute_import
  17 import unittest
  18 import mock
  19
  20 from oslo_serialization import jsonutils
  21
  22 from yardstick.network_services.traffic_profile import http_ixload
  23 from yardstick.network_services.traffic_profile.http_ixload import \
  24     join_non_strings, validate_non_string_sequence
  25
  26
  27 class TestJoinNonStrings(unittest.TestCase):
  28
  29     def test_validate_non_string_sequence(self):
  30         self.assertEqual(validate_non_string_sequence([1, 2, 3]), [1, 2, 3])
  31         self.assertIsNone(validate_non_string_sequence('123'))
  32         self.assertIsNone(validate_non_string_sequence(1))
  33
  34         self.assertEqual(validate_non_string_sequence(1, 2), 2)
  35         self.assertEqual(validate_non_string_sequence(1, default=2), 2)
  36
  37         with self.assertRaises(RuntimeError):
  38             validate_non_string_sequence(1, raise_exc=RuntimeError)
  39
  40     def test_join_non_strings(self):
  41         self.assertEqual(join_non_strings(':'), '')
  42         self.assertEqual(join_non_strings(':', 'a'), 'a')
  43         self.assertEqual(join_non_strings(':', 'a', 2, 'c'), 'a:2:c')
  44         self.assertEqual(join_non_strings(':', ['a', 2, 'c']), 'a:2:c')
  45         self.assertEqual(join_non_strings(':', 'abc'), 'abc')
  46
  47
  48 class TestIxLoadTrafficGen(unittest.TestCase):
  49
  50     def test_parse_run_test(self):
  51         ports = [1, 2, 3]
  52         test_input = {
  53             "remote_server": "REMOTE_SERVER",
  54             "ixload_cfg": "IXLOAD_CFG",
  55             "result_dir": "RESULT_DIR",
  56             "ixia_chassis": "IXIA_CHASSIS",
  57             "IXIA": {
  58                 "card": "CARD",
  59                 "ports": ports,
  60             },
  61         }
  62         j = jsonutils.dump_as_bytes(test_input)
  63         ixload = http_ixload.IXLOADHttpTest(j)
  64         self.assertDictEqual(ixload.test_input, test_input)
  65         self.assertIsNone(ixload.parse_run_test())
  66         self.assertEqual(ixload.ports_to_reassign, [
  67             ["IXIA_CHASSIS", "CARD", 1],
  68             ["IXIA_CHASSIS", "CARD", 2],
  69             ["IXIA_CHASSIS", "CARD", 3],
  70         ])
  71
  72     def test_format_ports_for_reassignment(self):
  73         ports = [
  74             ["IXIA_CHASSIS", "CARD", 1],
  75             ["IXIA_CHASSIS", "CARD", 2],
  76             ["IXIA_CHASSIS", "CARD", 3],
  77         ]
  78         formatted = http_ixload.IXLOADHttpTest.format_ports_for_reassignment(ports)
  79         self.assertEqual(formatted, [
  80             "IXIA_CHASSIS;CARD;1",
  81             "IXIA_CHASSIS;CARD;2",
  82             "IXIA_CHASSIS;CARD;3",
  83         ])
  84
  85     def test_reassign_ports(self):
  86         ports = [1, 2, 3]
  87         test_input = {
  88             "remote_server": "REMOTE_SERVER",
  89             "ixload_cfg": "IXLOAD_CFG",
  90             "result_dir": "RESULT_DIR",
  91             "ixia_chassis": "IXIA_CHASSIS",
  92             "IXIA": {
  93                 "card": "CARD",
  94                 "ports": ports,
  95             },
  96         }
  97         j = jsonutils.dump_as_bytes(test_input)
  98         ixload = http_ixload.IXLOADHttpTest(j)
  99         repository = mock.Mock()
 100         test = mock.MagicMock()
 101         test.setPorts = mock.Mock()
 102         ports_to_reassign = [(1, 2, 3), (1, 2, 4)]
 103         ixload.format_ports_for_reassignment = mock.Mock(return_value=["1;2;3"])
 104         self.assertIsNone(ixload.reassign_ports(test, repository, ports_to_reassign))
 105
 106     def test_reassign_ports_error(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         }
 118         j = jsonutils.dump_as_bytes(test_input)
 119         ixload = http_ixload.IXLOADHttpTest(j)
 120         repository = mock.Mock()
 121         test = "test"
 122         ports_to_reassign = [(1, 2, 3), (1, 2, 4)]
 123         ixload.format_ports_for_reassignment = mock.Mock(return_value=["1;2;3"])
 124         ixload.ix_load = mock.MagicMock()
 125         ixload.ix_load.delete = mock.Mock()
 126         ixload.ix_load.disconnect = mock.Mock()
 127         with self.assertRaises(Exception):
 128             ixload.reassign_ports(test, repository, ports_to_reassign)
 129
 130     def test_stat_collector(self):
 131         args = [0, 1]
 132         self.assertIsNone(
 133             http_ixload.IXLOADHttpTest.stat_collector(*args))
 134
 135     def test_IxL_StatCollectorCommand(self):
 136         args = [[0, 1, 2, 3], [0, 1, 2, 3]]
 137         self.assertIsNone(
 138             http_ixload.IXLOADHttpTest.IxL_StatCollectorCommand(*args))
 139
 140     def test_set_results_dir(self):
 141         test_stat_collector = mock.MagicMock()
 142         test_stat_collector.setResultDir = mock.Mock()
 143         results_on_windows = "c:/Results"
 144         self.assertIsNone(
 145             http_ixload.IXLOADHttpTest.set_results_dir(test_stat_collector,
 146                                                        results_on_windows))
 147
 148     def test_set_results_dir_error(self):
 149         test_stat_collector = ""
 150         results_on_windows = "c:/Results"
 151         with self.assertRaises(Exception):
 152             http_ixload.IXLOADHttpTest.set_results_dir(test_stat_collector, results_on_windows)
 153
 154     def test_load_config_file(self):
 155         ports = [1, 2, 3]
 156         test_input = {
 157             "remote_server": "REMOTE_SERVER",
 158             "ixload_cfg": "IXLOAD_CFG",
 159             "result_dir": "RESULT_DIR",
 160             "ixia_chassis": "IXIA_CHASSIS",
 161             "IXIA": {
 162                 "card": "CARD",
 163                 "ports": ports,
 164             },
 165         }
 166         j = jsonutils.dump_as_bytes(test_input)
 167         ixload = http_ixload.IXLOADHttpTest(j)
 168         ixload.ix_load = mock.MagicMock()
 169         ixload.ix_load.new = mock.Mock(return_value="")
 170         self.assertIsNotNone(ixload.load_config_file("ixload.cfg"))
 171
 172     def test_load_config_file_error(self):
 173         ports = [1, 2, 3]
 174         test_input = {
 175             "remote_server": "REMOTE_SERVER",
 176             "ixload_cfg": "IXLOAD_CFG",
 177             "result_dir": "RESULT_DIR",
 178             "ixia_chassis": "IXIA_CHASSIS",
 179             "IXIA": {
 180                 "card": "CARD",
 181                 "ports": ports,
 182             },
 183         }
 184         j = jsonutils.dump_as_bytes(test_input)
 185         ixload = http_ixload.IXLOADHttpTest(j)
 186         ixload.ix_load = "test"
 187         with self.assertRaises(Exception):
 188             ixload.load_config_file("ixload.cfg")
 189
 190     @mock.patch('yardstick.network_services.traffic_profile.http_ixload.IxLoad')
 191     @mock.patch('yardstick.network_services.traffic_profile.http_ixload.StatCollectorUtils')
 192     def test_start_http_test_connect_error(self, mock_collector_type, mock_ixload_type):
 193         ports = [1, 2, 3]
 194         test_input = {
 195             "remote_server": "REMOTE_SERVER",
 196             "ixload_cfg": "IXLOAD_CFG",
 197             "result_dir": "RESULT_DIR",
 198             "ixia_chassis": "IXIA_CHASSIS",
 199             "IXIA": {
 200                 "card": "CARD",
 201                 "ports": ports,
 202             },
 203         }
 204
 205         j = jsonutils.dump_as_bytes(test_input)
 206
 207         mock_ixload = mock_ixload_type()
 208         mock_ixload.connect.side_effect = RuntimeError
 209
 210         ixload = http_ixload.IXLOADHttpTest(j)
 211         ixload.results_on_windows = 'windows_result_dir'
 212         ixload.result_dir = 'my_result_dir'
 213
 214         with self.assertRaises(RuntimeError):
 215             ixload.start_http_test()
 216
 217     @mock.patch('yardstick.network_services.traffic_profile.http_ixload.IxLoad')
 218     @mock.patch('yardstick.network_services.traffic_profile.http_ixload.StatCollectorUtils')
 219     def test_start_http_test(self, mock_collector_type, mock_ixload_type):
 220         ports = [1, 2, 3]
 221         test_input = {
 222             "remote_server": "REMOTE_SERVER",
 223             "ixload_cfg": "IXLOAD_CFG",
 224             "result_dir": "RESULT_DIR",
 225             "ixia_chassis": "IXIA_CHASSIS",
 226             "IXIA": {
 227                 "card": "CARD",
 228                 "ports": ports,
 229             },
 230         }
 231
 232         j = jsonutils.dump_as_bytes(test_input)
 233
 234         ixload = http_ixload.IXLOADHttpTest(j)
 235         ixload.results_on_windows = 'windows_result_dir'
 236         ixload.result_dir = 'my_result_dir'
 237         ixload.load_config_file = mock.MagicMock()
 238
 239         self.assertIsNone(ixload.start_http_test())
 240
 241     @mock.patch('yardstick.network_services.traffic_profile.http_ixload.IxLoad')
 242     @mock.patch('yardstick.network_services.traffic_profile.http_ixload.StatCollectorUtils')
 243     def test_start_http_test_reassign_error(self, mock_collector_type, mock_ixload_type):
 244         ports = [1, 2, 3]
 245         test_input = {
 246             "remote_server": "REMOTE_SERVER",
 247             "ixload_cfg": "IXLOAD_CFG",
 248             "result_dir": "RESULT_DIR",
 249             "ixia_chassis": "IXIA_CHASSIS",
 250             "IXIA": {
 251                 "card": "CARD",
 252                 "ports": ports,
 253             },
 254         }
 255
 256         j = jsonutils.dump_as_bytes(test_input)
 257
 258         ixload = http_ixload.IXLOADHttpTest(j)
 259         ixload.load_config_file = mock.MagicMock()
 260
 261         reassign_ports = mock.Mock(side_effect=RuntimeError)
 262         ixload.reassign_ports = reassign_ports
 263         ixload.results_on_windows = 'windows_result_dir'
 264         ixload.result_dir = 'my_result_dir'
 265
 266         ixload.start_http_test()
 267         self.assertEqual(reassign_ports.call_count, 1)
 268
 269     @mock.patch("yardstick.network_services.traffic_profile.http_ixload.IXLOADHttpTest")
 270     def test_main(self, IXLOADHttpTest):
 271         args = ["1", "2", "3"]
 272         http_ixload.main(args)