yardstick/tests/unit/network_services/traffic_profile/test_prox_binsearch.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 yardstick.network_services.vnf_generic.vnf.prox_helpers import ProxTestDataTuple
  19 from yardstick.network_services.traffic_profile import prox_binsearch
  20
  21
  22 class TestProxBinSearchProfile(unittest.TestCase):
  23
  24     THEOR_MAX_THROUGHPUT = 0.00012340000000000002
  25
  26     def setUp(self):
  27         self._mock_log_info = mock.patch.object(prox_binsearch.LOG, 'info')
  28         self.mock_log_info = self._mock_log_info.start()
  29         self.addCleanup(self._stop_mocks)
  30
  31     def _stop_mocks(self):
  32         self._mock_log_info.stop()
  33
  34     def test_execute_1(self):
  35         def target(*args, **_):
  36             runs.append(args[2])
  37             if args[2] < 0 or args[2] > 100:
  38                 raise RuntimeError(' '.join([str(args), str(runs)]))
  39             if args[2] > 75.0:
  40                 return fail_tuple, {}
  41             return success_tuple, {}
  42
  43         def side_effect_func(arg1, arg2):
  44             if arg1 == "confirmation":
  45                 return arg2
  46             else:
  47                 return {}
  48
  49         tp_config = {
  50             'traffic_profile': {
  51                 'packet_sizes': [200],
  52                 'test_precision': 2.0,
  53                 'tolerated_loss': 0.001,
  54             },
  55         }
  56
  57         runs = []
  58         success_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.1, 5.2, 5.3], 995, 1000, 123.4)
  59         fail_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.6, 5.7, 5.8], 850, 1000, 123.4)
  60
  61         traffic_generator = mock.MagicMock()
  62         attrs1 = {'get.return_value': 10}
  63         traffic_generator.scenario_helper.all_options.configure_mock(**attrs1)
  64
  65         attrs2 = {'__getitem__.return_value': 10, 'get.return_value': 10}
  66         attrs3 = {'get.side_effect': side_effect_func}
  67         traffic_generator.scenario_helper.scenario_cfg["runner"].configure_mock(**attrs2)
  68         traffic_generator.scenario_helper.scenario_cfg["options"].configure_mock(**attrs3)
  69
  70         profile_helper = mock.MagicMock()
  71         profile_helper.run_test = target
  72
  73         profile = prox_binsearch.ProxBinSearchProfile(tp_config)
  74         profile.init(mock.MagicMock())
  75         profile._profile_helper = profile_helper
  76
  77         profile.execute_traffic(traffic_generator)
  78
  79         self.assertEqual(round(profile.current_lower, 2), 74.69)
  80         self.assertEqual(round(profile.current_upper, 2), 76.09)
  81         self.assertEqual(len(runs), 7)
  82
  83         # Result Samples inc theor_max
  84         result_tuple = {'Actual_throughput': 5e-07,
  85                         'theor_max_throughput': self.THEOR_MAX_THROUGHPUT,
  86                         'PktSize': 200,
  87                         'Status': 'Result'}
  88
  89         test_results = profile.queue.put.call_args[0]
  90         for k in result_tuple:
  91             self.assertEqual(result_tuple[k], test_results[0][k])
  92
  93         success_result_tuple = {"CurrentDropPackets": 0.5,
  94                                 "DropPackets": 0.5,
  95                                 "LatencyAvg": 5.3,
  96                                 "LatencyMax": 5.2,
  97                                 "LatencyMin": 5.1,
  98                                 "PktSize": 200,
  99                                 "RxThroughput": 7.5e-07,
 100                                 "Throughput": 7.5e-07,
 101                                 "TxThroughput": self.THEOR_MAX_THROUGHPUT,
 102                                 "Status": 'Success'}
 103
 104         calls = profile.queue.put(success_result_tuple)
 105         profile.queue.put.assert_has_calls(calls)
 106
 107         success_result_tuple2 = {"CurrentDropPackets": 0.5,
 108                                 "DropPackets": 0.5,
 109                                 "LatencyAvg": 5.3,
 110                                 "LatencyMax": 5.2,
 111                                 "LatencyMin": 5.1,
 112                                 "PktSize": 200,
 113                                 "RxThroughput": 7.5e-07,
 114                                 "Throughput": 7.5e-07,
 115                                 "TxThroughput": 123.4,
 116                                 "can_be_lost": 409600,
 117                                 "drop_total": 20480,
 118                                 "rx_total": 4075520,
 119                                 "tx_total": 4096000,
 120                                 "Status": 'Success'}
 121
 122         calls = profile.queue.put(success_result_tuple2)
 123         profile.queue.put.assert_has_calls(calls)
 124
 125     def test_execute_2(self):
 126         def target(*args, **_):
 127             runs.append(args[2])
 128             if args[2] < 0 or args[2] > 100:
 129                 raise RuntimeError(' '.join([str(args), str(runs)]))
 130             if args[2] > 25.0:
 131                 return fail_tuple, {}
 132             return success_tuple, {}
 133
 134         def side_effect_func(arg1, _):
 135             if arg1 == "confirmation":
 136                 return 2
 137             else:
 138                 return {}
 139
 140         tp_config = {
 141             'traffic_profile': {
 142                 'packet_sizes': [200],
 143                 'test_precision': 2.0,
 144                 'tolerated_loss': 0.001,
 145             },
 146         }
 147
 148         runs = []
 149         success_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.1, 5.2, 5.3], 995, 1000, 123.4)
 150         fail_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.6, 5.7, 5.8], 850, 1000, 123.4)
 151
 152         traffic_generator = mock.MagicMock()
 153         attrs1 = {'get.return_value': 10}
 154         traffic_generator.scenario_helper.all_options.configure_mock(**attrs1)
 155
 156         attrs2 = {'__getitem__.return_value': 0, 'get.return_value': 0}
 157         attrs3 = {'get.side_effect': side_effect_func}
 158
 159         traffic_generator.scenario_helper.scenario_cfg["runner"].configure_mock(**attrs2)
 160         traffic_generator.scenario_helper.scenario_cfg["options"].configure_mock(**attrs3)
 161
 162         profile_helper = mock.MagicMock()
 163         profile_helper.run_test = target
 164
 165         profile = prox_binsearch.ProxBinSearchProfile(tp_config)
 166         profile.init(mock.MagicMock())
 167         profile._profile_helper = profile_helper
 168
 169         profile.execute_traffic(traffic_generator)
 170         self.assertEqual(round(profile.current_lower, 2), 24.06)
 171         self.assertEqual(round(profile.current_upper, 2), 25.47)
 172         self.assertEqual(len(runs), 21)
 173
 174     def test_execute_3(self):
 175         def target(*args, **_):
 176             runs.append(args[2])
 177             if args[2] < 0 or args[2] > 100:
 178                 raise RuntimeError(' '.join([str(args), str(runs)]))
 179             if args[2] > 75.0:
 180                 return fail_tuple, {}
 181             return success_tuple, {}
 182
 183         tp_config = {
 184             'traffic_profile': {
 185                 'packet_sizes': [200],
 186                 'test_precision': 2.0,
 187                 'tolerated_loss': 0.001,
 188             },
 189         }
 190
 191         runs = []
 192         success_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.1, 5.2, 5.3], 995, 1000, 123.4)
 193         fail_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.6, 5.7, 5.8], 850, 1000, 123.4)
 194
 195         traffic_generator = mock.MagicMock()
 196
 197         profile_helper = mock.MagicMock()
 198         profile_helper.run_test = target
 199
 200         profile = prox_binsearch.ProxBinSearchProfile(tp_config)
 201         profile.init(mock.MagicMock())
 202         profile._profile_helper = profile_helper
 203
 204         profile.upper_bound = 100.0
 205         profile.lower_bound = 99.0
 206         profile.execute_traffic(traffic_generator)
 207
 208         result_tuple = {'Actual_throughput': 0, 'theor_max_throughput': 0,
 209                         "Status": 'Result', "Next_Step": ''}
 210         profile.queue.put.assert_called_with(result_tuple)
 211
 212         # Check for success_ tuple (None expected)
 213         calls = profile.queue.put.mock_calls
 214         for call in calls:
 215             for call_detail in call[1]:
 216                 if call_detail["Status"] == 'Success':
 217                     self.assertRaises(AttributeError)
 218
 219     def test_execute_4(self):
 220
 221         def target(*args, **_):
 222             runs.append(args[2])
 223             if args[2] < 0 or args[2] > 100:
 224                 raise RuntimeError(' '.join([str(args), str(runs)]))
 225             if args[2] > 75.0:
 226                 return fail_tuple, {}
 227
 228             return success_tuple, {}
 229
 230         tp_config = {
 231             'traffic_profile': {
 232                 'packet_sizes': [200],
 233                 'test_precision': 2.0,
 234                 'tolerated_loss': 0.001,
 235             },
 236         }
 237
 238         runs = []
 239         success_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.1, 5.2, 5.3], 995, 1000, 123.4)
 240         fail_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.6, 5.7, 5.8], 850, 1000, 123.4)
 241
 242         traffic_generator = mock.MagicMock()
 243         attrs1 = {'get.return_value': 100000}
 244         traffic_generator.scenario_helper.all_options.configure_mock(**attrs1)
 245
 246         attrs2 = {'__getitem__.return_value': 0, 'get.return_value': 0}
 247
 248         traffic_generator.scenario_helper.scenario_cfg["runner"].configure_mock(**attrs2)
 249
 250         profile_helper = mock.MagicMock()
 251         profile_helper.run_test = target
 252
 253         profile = prox_binsearch.ProxBinSearchProfile(tp_config)
 254         profile.init(mock.MagicMock())
 255         profile._profile_helper = profile_helper
 256
 257         profile.execute_traffic(traffic_generator)
 258         self.assertEqual(round(profile.current_lower, 2), 74.69)
 259         self.assertEqual(round(profile.current_upper, 2), 76.09)
 260         self.assertEqual(len(runs), 7)
 261
 262         # Result Samples inc theor_max
 263         result_tuple = {'Actual_throughput': 5e-07,
 264                         'theor_max_throughput': self.THEOR_MAX_THROUGHPUT,
 265                         'PktSize': 200,
 266                         "Status": 'Result'}
 267
 268         test_results = profile.queue.put.call_args[0]
 269         for k in result_tuple:
 270             self.assertEqual(result_tuple[k], test_results[0][k])
 271
 272         success_result_tuple = {"CurrentDropPackets": 0.5,
 273                                 "DropPackets": 0.5,
 274                                 "LatencyAvg": 5.3,
 275                                 "LatencyMax": 5.2,
 276                                 "LatencyMin": 5.1,
 277                                 "PktSize": 200,
 278                                 "RxThroughput": 7.5e-07,
 279                                 "Throughput": 7.5e-07,
 280                                 "TxThroughput": self.THEOR_MAX_THROUGHPUT,
 281                                 "Status": 'Success'}
 282
 283         calls = profile.queue.put(success_result_tuple)
 284         profile.queue.put.assert_has_calls(calls)
 285
 286         success_result_tuple2 = {"CurrentDropPackets": 0.5,
 287                                  "DropPackets": 0.5,
 288                                  "LatencyAvg": 5.3,
 289                                  "LatencyMax": 5.2,
 290                                  "LatencyMin": 5.1,
 291                                  "PktSize": 200,
 292                                  "RxThroughput": 7.5e-07,
 293                                  "Throughput": 7.5e-07,
 294                                  "TxThroughput": 123.4,
 295                                  "can_be_lost": 409600,
 296                                  "drop_total": 20480,
 297                                  "rx_total": 4075520,
 298                                  "tx_total": 4096000,
 299                                  "Status": 'Success'}
 300
 301         calls = profile.queue.put(success_result_tuple2)
 302         profile.queue.put.assert_has_calls(calls)