1 # Copyright (c) 2017 Intel Corporation
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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.
18 from yardstick.network_services.vnf_generic.vnf.prox_helpers import ProxTestDataTuple
19 from yardstick.network_services.traffic_profile import prox_binsearch
22 class TestProxBinSearchProfile(unittest.TestCase):
25 self._mock_log_info = mock.patch.object(prox_binsearch.LOG, 'info')
26 self.mock_log_info = self._mock_log_info.start()
27 self.addCleanup(self._stop_mocks)
29 def _stop_mocks(self):
30 self._mock_log_info.stop()
32 def test_execute_1(self):
33 def target(*args, **_):
35 if args[2] < 0 or args[2] > 100:
36 raise RuntimeError(' '.join([str(args), str(runs)]))
39 return success_tuple, {}
41 def side_effect_func(arg1, arg2):
42 if arg1 == "confirmation":
49 'packet_sizes': [200],
50 'test_precision': 2.0,
51 'tolerated_loss': 0.001,
56 success_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.1, 5.2, 5.3], 995, 1000, 123.4)
57 fail_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.6, 5.7, 5.8], 850, 1000, 123.4)
59 traffic_generator = mock.MagicMock()
60 attrs1 = {'get.return_value': 10}
61 traffic_generator.scenario_helper.all_options.configure_mock(**attrs1)
63 attrs2 = {'__getitem__.return_value': 10, 'get.return_value': 10}
64 attrs3 = {'get.side_effect': side_effect_func}
65 traffic_generator.scenario_helper.scenario_cfg["runner"].configure_mock(**attrs2)
66 traffic_generator.scenario_helper.scenario_cfg["options"].configure_mock(**attrs3)
68 profile_helper = mock.MagicMock()
69 profile_helper.run_test = target
71 profile = prox_binsearch.ProxBinSearchProfile(tp_config)
72 profile.init(mock.MagicMock())
73 profile._profile_helper = profile_helper
75 profile.execute_traffic(traffic_generator)
77 self.assertEqual(round(profile.current_lower, 2), 74.69)
78 self.assertEqual(round(profile.current_upper, 2), 76.09)
79 self.assertEqual(len(runs), 7)
81 # Result Samples inc theor_max
82 result_tuple = {'Actual_throughput': 5e-07,
83 'theor_max_throughput': 7.5e-07,
87 test_results = profile.queue.put.call_args[0]
88 for k in result_tuple:
89 self.assertEqual(result_tuple[k], test_results[0][k])
91 success_result_tuple = {"CurrentDropPackets": 0.5,
97 "RxThroughput": 7.5e-07,
98 "Throughput": 7.5e-07,
99 "TxThroughput": 0.00012340000000000002,
102 calls = profile.queue.put(success_result_tuple)
103 profile.queue.put.assert_has_calls(calls)
105 success_result_tuple2 = {"CurrentDropPackets": 0.5,
111 "RxThroughput": 7.5e-07,
112 "Throughput": 7.5e-07,
113 "TxThroughput": 123.4,
114 "can_be_lost": 409600,
120 calls = profile.queue.put(success_result_tuple2)
121 profile.queue.put.assert_has_calls(calls)
123 def test_execute_2(self):
124 def target(*args, **_):
126 if args[2] < 0 or args[2] > 100:
127 raise RuntimeError(' '.join([str(args), str(runs)]))
129 return fail_tuple, {}
130 return success_tuple, {}
132 def side_effect_func(arg1, _):
133 if arg1 == "confirmation":
140 'packet_sizes': [200],
141 'test_precision': 2.0,
142 'tolerated_loss': 0.001,
147 success_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.1, 5.2, 5.3], 995, 1000, 123.4)
148 fail_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.6, 5.7, 5.8], 850, 1000, 123.4)
150 traffic_generator = mock.MagicMock()
151 attrs1 = {'get.return_value': 10}
152 traffic_generator.scenario_helper.all_options.configure_mock(**attrs1)
154 attrs2 = {'__getitem__.return_value': 0, 'get.return_value': 0}
155 attrs3 = {'get.side_effect': side_effect_func}
157 traffic_generator.scenario_helper.scenario_cfg["runner"].configure_mock(**attrs2)
158 traffic_generator.scenario_helper.scenario_cfg["options"].configure_mock(**attrs3)
160 profile_helper = mock.MagicMock()
161 profile_helper.run_test = target
163 profile = prox_binsearch.ProxBinSearchProfile(tp_config)
164 profile.init(mock.MagicMock())
165 profile._profile_helper = profile_helper
167 profile.execute_traffic(traffic_generator)
168 self.assertEqual(round(profile.current_lower, 2), 24.06)
169 self.assertEqual(round(profile.current_upper, 2), 25.47)
170 self.assertEqual(len(runs), 21)
172 def test_execute_3(self):
173 def target(*args, **_):
175 if args[2] < 0 or args[2] > 100:
176 raise RuntimeError(' '.join([str(args), str(runs)]))
178 return fail_tuple, {}
179 return success_tuple, {}
183 'packet_sizes': [200],
184 'test_precision': 2.0,
185 'tolerated_loss': 0.001,
190 success_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.1, 5.2, 5.3], 995, 1000, 123.4)
191 fail_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.6, 5.7, 5.8], 850, 1000, 123.4)
193 traffic_generator = mock.MagicMock()
195 profile_helper = mock.MagicMock()
196 profile_helper.run_test = target
198 profile = prox_binsearch.ProxBinSearchProfile(tp_config)
199 profile.init(mock.MagicMock())
200 profile._profile_helper = profile_helper
202 profile.upper_bound = 100.0
203 profile.lower_bound = 99.0
204 profile.execute_traffic(traffic_generator)
206 result_tuple = {'Actual_throughput': 0, 'theor_max_throughput': 0,
207 "Status": 'Result', "Next_Step": ''}
208 profile.queue.put.assert_called_with(result_tuple)
210 # Check for success_ tuple (None expected)
211 calls = profile.queue.put.mock_calls
213 for call_detail in call[1]:
214 if call_detail["Status"] == 'Success':
215 self.assertRaises(AttributeError)
217 def test_execute_4(self):
219 def target(*args, **_):
221 if args[2] < 0 or args[2] > 100:
222 raise RuntimeError(' '.join([str(args), str(runs)]))
224 return fail_tuple, {}
225 return success_tuple, {}
229 'packet_sizes': [200],
230 'test_precision': 2.0,
231 'tolerated_loss': 0.001,
236 success_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.1, 5.2, 5.3], 995, 1000, 123.4)
237 fail_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.6, 5.7, 5.8], 850, 1000, 123.4)
239 traffic_generator = mock.MagicMock()
240 attrs1 = {'get.return_value': 100000}
241 traffic_generator.scenario_helper.all_options.configure_mock(**attrs1)
243 attrs2 = {'__getitem__.return_value': 0, 'get.return_value': 0}
245 traffic_generator.scenario_helper.scenario_cfg["runner"].configure_mock(**attrs2)
247 profile_helper = mock.MagicMock()
248 profile_helper.run_test = target
250 profile = prox_binsearch.ProxBinSearchProfile(tp_config)
251 profile.init(mock.MagicMock())
252 profile._profile_helper = profile_helper
254 profile.execute_traffic(traffic_generator)
255 self.assertEqual(round(profile.current_lower, 2), 74.69)
256 self.assertEqual(round(profile.current_upper, 2), 76.09)
257 self.assertEqual(len(runs), 7)
259 # Result Samples inc theor_max
260 result_tuple = {'Actual_throughput': 5e-07,
261 'theor_max_throughput': 7.5e-07,
265 test_results = profile.queue.put.call_args[0]
266 for k in result_tuple:
267 self.assertEqual(result_tuple[k], test_results[0][k])
269 success_result_tuple = {"CurrentDropPackets": 0.5,
275 "RxThroughput": 7.5e-07,
276 "Throughput": 7.5e-07,
277 "TxThroughput": 0.00012340000000000002,
280 calls = profile.queue.put(success_result_tuple)
281 profile.queue.put.assert_has_calls(calls)
283 success_result_tuple2 = {"CurrentDropPackets": 0.5,
289 "RxThroughput": 7.5e-07,
290 "Throughput": 7.5e-07,
291 "TxThroughput": 123.4,
292 "can_be_lost": 409600,
298 calls = profile.queue.put(success_result_tuple2)
299 profile.queue.put.assert_has_calls(calls)