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, {}
43 'packet_sizes': [200],
44 'test_precision': 2.0,
45 'tolerated_loss': 0.001,
50 success_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.1, 5.2, 5.3], 995, 1000, 123.4)
51 fail_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.6, 5.7, 5.8], 850, 1000, 123.4)
53 traffic_generator = mock.MagicMock()
54 attrs1 = {'get.return_value' : 10}
55 traffic_generator.scenario_helper.all_options.configure_mock(**attrs1)
57 attrs2 = {'__getitem__.return_value' : 10, 'get.return_value': 10}
58 traffic_generator.scenario_helper.scenario_cfg["runner"].configure_mock(**attrs2)
60 profile_helper = mock.MagicMock()
61 profile_helper.run_test = target
63 profile = prox_binsearch.ProxBinSearchProfile(tp_config)
64 profile.init(mock.MagicMock())
65 profile._profile_helper = profile_helper
67 profile.execute_traffic(traffic_generator)
69 self.assertEqual(round(profile.current_lower, 2), 74.69)
70 self.assertEqual(round(profile.current_upper, 2), 76.09)
71 self.assertEqual(len(runs), 77)
73 # Result Samples inc theor_max
74 result_tuple = {'Actual_throughput': 5e-07,
75 'theor_max_throughput': 7.5e-07,
79 test_results = profile.queue.put.call_args[0]
80 for k in result_tuple:
81 self.assertEqual(result_tuple[k], test_results[0][k])
83 success_result_tuple = {"CurrentDropPackets": 0.5,
89 "RxThroughput": 7.5e-07,
90 "Throughput": 7.5e-07,
91 "TxThroughput": 0.00012340000000000002,
94 calls = profile.queue.put(success_result_tuple)
95 profile.queue.put.assert_has_calls(calls)
97 success_result_tuple2 = {"CurrentDropPackets": 0.5,
103 "RxThroughput": 7.5e-07,
104 "Throughput": 7.5e-07,
105 "TxThroughput": 123.4,
106 "can_be_lost": 409600,
112 calls = profile.queue.put(success_result_tuple2)
113 profile.queue.put.assert_has_calls(calls)
115 def test_execute_2(self):
116 def target(*args, **_):
118 if args[2] < 0 or args[2] > 100:
119 raise RuntimeError(' '.join([str(args), str(runs)]))
121 return fail_tuple, {}
122 return success_tuple, {}
126 'packet_sizes': [200],
127 'test_precision': 2.0,
128 'tolerated_loss': 0.001,
133 success_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.1, 5.2, 5.3], 995, 1000, 123.4)
134 fail_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.6, 5.7, 5.8], 850, 1000, 123.4)
136 traffic_generator = mock.MagicMock()
137 attrs1 = {'get.return_value': 10}
138 traffic_generator.scenario_helper.all_options.configure_mock(**attrs1)
140 attrs2 = {'__getitem__.return_value': 0, 'get.return_value': 0}
141 traffic_generator.scenario_helper.scenario_cfg["runner"].configure_mock(**attrs2)
143 profile_helper = mock.MagicMock()
144 profile_helper.run_test = target
146 profile = prox_binsearch.ProxBinSearchProfile(tp_config)
147 profile.init(mock.MagicMock())
148 profile._profile_helper = profile_helper
150 profile.execute_traffic(traffic_generator)
151 self.assertEqual(round(profile.current_lower, 2), 24.06)
152 self.assertEqual(round(profile.current_upper, 2), 25.47)
153 self.assertEqual(len(runs), 7)
155 def test_execute_3(self):
156 def target(*args, **_):
158 if args[2] < 0 or args[2] > 100:
159 raise RuntimeError(' '.join([str(args), str(runs)]))
161 return fail_tuple, {}
162 return success_tuple, {}
166 'packet_sizes': [200],
167 'test_precision': 2.0,
168 'tolerated_loss': 0.001,
173 success_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.1, 5.2, 5.3], 995, 1000, 123.4)
174 fail_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.6, 5.7, 5.8], 850, 1000, 123.4)
176 traffic_generator = mock.MagicMock()
178 profile_helper = mock.MagicMock()
179 profile_helper.run_test = target
181 profile = prox_binsearch.ProxBinSearchProfile(tp_config)
182 profile.init(mock.MagicMock())
183 profile._profile_helper = profile_helper
185 profile.upper_bound = 100.0
186 profile.lower_bound = 99.0
187 profile.execute_traffic(traffic_generator)
191 result_tuple = {'Actual_throughput': 0, 'theor_max_throughput': 0,
192 "Status": 'Result', "Next_Step": ''}
193 profile.queue.put.assert_called_with(result_tuple)
195 # Check for success_ tuple (None expected)
196 calls = profile.queue.put.mock_calls
198 for call_detail in call[1]:
199 if call_detail["Status"] == 'Success':
200 self.assertRaises(AttributeError)
202 def test_execute_4(self):
204 def target(*args, **_):
206 if args[2] < 0 or args[2] > 100:
207 raise RuntimeError(' '.join([str(args), str(runs)]))
209 return fail_tuple, {}
210 return success_tuple, {}
214 'packet_sizes': [200],
215 'test_precision': 2.0,
216 'tolerated_loss': 0.001,
221 success_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.1, 5.2, 5.3], 995, 1000, 123.4)
222 fail_tuple = ProxTestDataTuple(10.0, 1, 2, 3, 4, [5.6, 5.7, 5.8], 850, 1000, 123.4)
224 traffic_generator = mock.MagicMock()
225 attrs1 = {'get.return_value': 100000}
226 traffic_generator.scenario_helper.all_options.configure_mock(**attrs1)
228 attrs2 = {'__getitem__.return_value': 0, 'get.return_value': 0}
229 traffic_generator.scenario_helper.scenario_cfg["runner"].configure_mock(**attrs2)
231 profile_helper = mock.MagicMock()
232 profile_helper.run_test = target
234 profile = prox_binsearch.ProxBinSearchProfile(tp_config)
235 profile.init(mock.MagicMock())
236 profile._profile_helper = profile_helper
238 profile.execute_traffic(traffic_generator)
239 self.assertEqual(round(profile.current_lower, 2), 74.69)
240 self.assertEqual(round(profile.current_upper, 2), 76.09)
241 self.assertEqual(len(runs), 7)
243 # Result Samples inc theor_max
244 result_tuple = {'Actual_throughput': 5e-07,
245 'theor_max_throughput': 7.5e-07,
249 test_results = profile.queue.put.call_args[0]
250 for k in result_tuple:
251 self.assertEqual(result_tuple[k], test_results[0][k])
253 success_result_tuple = {"CurrentDropPackets": 0.5,
259 "RxThroughput": 7.5e-07,
260 "Throughput": 7.5e-07,
261 "TxThroughput": 0.00012340000000000002,
264 calls = profile.queue.put(success_result_tuple)
265 profile.queue.put.assert_has_calls(calls)
267 success_result_tuple2 = {"CurrentDropPackets": 0.5,
273 "RxThroughput": 7.5e-07,
274 "Throughput": 7.5e-07,
275 "TxThroughput": 123.4,
276 "can_be_lost": 409600,
282 calls = profile.queue.put(success_result_tuple2)
283 profile.queue.put.assert_has_calls(calls)