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.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 test_parse_run_test(self):
  49         ports = [1, 2, 3]
  50         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         }
  60         j = jsonutils.dump_as_bytes(test_input)
  61         ixload = http_ixload.IXLOADHttpTest(j)
  62         self.assertDictEqual(ixload.test_input, test_input)
  63         self.assertIsNone(ixload.parse_run_test())
  64         self.assertEqual(ixload.ports_to_reassign, [
  65             ["IXIA_CHASSIS", "CARD", 1],
  66             ["IXIA_CHASSIS", "CARD", 2],
  67             ["IXIA_CHASSIS", "CARD", 3],
  68         ])
  69
  70     def test_format_ports_for_reassignment(self):
  71         ports = [
  72             ["IXIA_CHASSIS", "CARD", 1],
  73             ["IXIA_CHASSIS", "CARD", 2],
  74             ["IXIA_CHASSIS", "CARD", 3],
  75         ]
  76         formatted = http_ixload.IXLOADHttpTest.format_ports_for_reassignment(ports)
  77         self.assertEqual(formatted, [
  78             "IXIA_CHASSIS;CARD;1",
  79             "IXIA_CHASSIS;CARD;2",
  80             "IXIA_CHASSIS;CARD;3",
  81         ])
  82
  83     def test_reassign_ports(self):
  84         ports = [1, 2, 3]
  85         test_input = {
  86             "remote_server": "REMOTE_SERVER",
  87             "ixload_cfg": "IXLOAD_CFG",
  88             "result_dir": "RESULT_DIR",
  89             "ixia_chassis": "IXIA_CHASSIS",
  90             "IXIA": {
  91                 "card": "CARD",
  92                 "ports": ports,
  93             },
  94         }
  95         j = jsonutils.dump_as_bytes(test_input)
  96         ixload = http_ixload.IXLOADHttpTest(j)
  97         repository = mock.Mock()
  98         test = mock.MagicMock()
  99         test.setPorts = mock.Mock()
 100         ports_to_reassign = [(1, 2, 3), (1, 2, 4)]
 101         ixload.format_ports_for_reassignment = mock.Mock(return_value=["1;2;3"])
 102         self.assertIsNone(ixload.reassign_ports(test, repository, ports_to_reassign))
 103
 104     def test_reassign_ports_error(self):
 105         ports = [1, 2, 3]
 106         test_input = {
 107             "remote_server": "REMOTE_SERVER",
 108             "ixload_cfg": "IXLOAD_CFG",
 109             "result_dir": "RESULT_DIR",
 110             "ixia_chassis": "IXIA_CHASSIS",
 111             "IXIA": {
 112                 "card": "CARD",
 113                 "ports": ports,
 114             },
 115         }
 116         j = jsonutils.dump_as_bytes(test_input)
 117         ixload = http_ixload.IXLOADHttpTest(j)
 118         repository = mock.Mock()
 119         test = "test"
 120         ports_to_reassign = [(1, 2, 3), (1, 2, 4)]
 121         ixload.format_ports_for_reassignment = mock.Mock(return_value=["1;2;3"])
 122         ixload.ix_load = mock.MagicMock()
 123         ixload.ix_load.delete = mock.Mock()
 124         ixload.ix_load.disconnect = mock.Mock()
 125         with self.assertRaises(Exception):
 126             ixload.reassign_ports(test, repository, ports_to_reassign)
 127
 128     def test_stat_collector(self):
 129         args = [0, 1]
 130         self.assertIsNone(
 131             http_ixload.IXLOADHttpTest.stat_collector(*args))
 132
 133     def test_IxL_StatCollectorCommand(self):
 134         args = [[0, 1, 2, 3], [0, 1, 2, 3]]
 135         self.assertIsNone(
 136             http_ixload.IXLOADHttpTest.IxL_StatCollectorCommand(*args))
 137
 138     def test_set_results_dir(self):
 139         test_stat_collector = mock.MagicMock()
 140         test_stat_collector.setResultDir = mock.Mock()
 141         results_on_windows = "c:/Results"
 142         self.assertIsNone(
 143             http_ixload.IXLOADHttpTest.set_results_dir(test_stat_collector,
 144                                                        results_on_windows))
 145
 146     def test_set_results_dir_error(self):
 147         test_stat_collector = ""
 148         results_on_windows = "c:/Results"
 149         with self.assertRaises(Exception):
 150             http_ixload.IXLOADHttpTest.set_results_dir(test_stat_collector, results_on_windows)
 151
 152     def test_load_config_file(self):
 153         ports = [1, 2, 3]
 154         test_input = {
 155             "remote_server": "REMOTE_SERVER",
 156             "ixload_cfg": "IXLOAD_CFG",
 157             "result_dir": "RESULT_DIR",
 158             "ixia_chassis": "IXIA_CHASSIS",
 159             "IXIA": {
 160                 "card": "CARD",
 161                 "ports": ports,
 162             },
 163         }
 164         j = jsonutils.dump_as_bytes(test_input)
 165         ixload = http_ixload.IXLOADHttpTest(j)
 166         ixload.ix_load = mock.MagicMock()
 167         ixload.ix_load.new = mock.Mock(return_value="")
 168         self.assertIsNotNone(ixload.load_config_file("ixload.cfg"))
 169
 170     def test_load_config_file_error(self):
 171         ports = [1, 2, 3]
 172         test_input = {
 173             "remote_server": "REMOTE_SERVER",
 174             "ixload_cfg": "IXLOAD_CFG",
 175             "result_dir": "RESULT_DIR",
 176             "ixia_chassis": "IXIA_CHASSIS",
 177             "IXIA": {
 178                 "card": "CARD",
 179                 "ports": ports,
 180             },
 181         }
 182         j = jsonutils.dump_as_bytes(test_input)
 183         ixload = http_ixload.IXLOADHttpTest(j)
 184         ixload.ix_load = "test"
 185         with self.assertRaises(Exception):
 186             ixload.load_config_file("ixload.cfg")
 187
 188     @mock.patch('yardstick.network_services.traffic_profile.http_ixload.StatCollectorUtils')
 189     @mock.patch('yardstick.network_services.traffic_profile.http_ixload.IxLoad')
 190     def test_start_http_test_connect_error(self, mock_ixload_type, *args):
 191         ports = [1, 2, 3]
 192         test_input = {
 193             "remote_server": "REMOTE_SERVER",
 194             "ixload_cfg": "IXLOAD_CFG",
 195             "result_dir": "RESULT_DIR",
 196             "ixia_chassis": "IXIA_CHASSIS",
 197             "IXIA": {
 198                 "card": "CARD",
 199                 "ports": ports,
 200             },
 201         }
 202
 203         j = jsonutils.dump_as_bytes(test_input)
 204
 205         mock_ixload_type.return_value.connect.side_effect = RuntimeError
 206
 207         ixload = http_ixload.IXLOADHttpTest(j)
 208         ixload.results_on_windows = 'windows_result_dir'
 209         ixload.result_dir = 'my_result_dir'
 210
 211         with self.assertRaises(RuntimeError):
 212             ixload.start_http_test()
 213
 214     @mock.patch('yardstick.network_services.traffic_profile.http_ixload.IxLoad')
 215     @mock.patch('yardstick.network_services.traffic_profile.http_ixload.StatCollectorUtils')
 216     def test_start_http_test(self, *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         }
 228
 229         j = jsonutils.dump_as_bytes(test_input)
 230
 231         ixload = http_ixload.IXLOADHttpTest(j)
 232         ixload.results_on_windows = 'windows_result_dir'
 233         ixload.result_dir = 'my_result_dir'
 234         ixload.load_config_file = mock.MagicMock()
 235
 236         self.assertIsNone(ixload.start_http_test())
 237
 238     @mock.patch('yardstick.network_services.traffic_profile.http_ixload.IxLoad')
 239     @mock.patch('yardstick.network_services.traffic_profile.http_ixload.StatCollectorUtils')
 240     def test_start_http_test_reassign_error(self, *args):
 241         ports = [1, 2, 3]
 242         test_input = {
 243             "remote_server": "REMOTE_SERVER",
 244             "ixload_cfg": "IXLOAD_CFG",
 245             "result_dir": "RESULT_DIR",
 246             "ixia_chassis": "IXIA_CHASSIS",
 247             "IXIA": {
 248                 "card": "CARD",
 249                 "ports": ports,
 250             },
 251         }
 252
 253         j = jsonutils.dump_as_bytes(test_input)
 254
 255         ixload = http_ixload.IXLOADHttpTest(j)
 256         ixload.load_config_file = mock.MagicMock()
 257
 258         reassign_ports = mock.Mock(side_effect=RuntimeError)
 259         ixload.reassign_ports = reassign_ports
 260         ixload.results_on_windows = 'windows_result_dir'
 261         ixload.result_dir = 'my_result_dir'
 262
 263         ixload.start_http_test()
 264         reassign_ports.assert_called_once()
 265
 266     @mock.patch("yardstick.network_services.traffic_profile.http_ixload.IXLOADHttpTest")
 267     def test_main(self, *args):
 268         args = ["1", "2", "3"]
 269         http_ixload.main(args)