1#!/usr/bin/env python3.4
2#
3#   Copyright 2017 - The Android Open Source Project
4#
5#   Licensed under the Apache License, Version 2.0 (the 'License');
6#   you may not use this file except in compliance with the License.
7#   You may obtain a copy of the License at
8#
9#       http://www.apache.org/licenses/LICENSE-2.0
10#
11#   Unless required by applicable law or agreed to in writing, software
12#   distributed under the License is distributed on an 'AS IS' BASIS,
13#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14#   See the License for the specific language governing permissions and
15#   limitations under the License.
16
17import collections
18import itertools
19import json
20import logging
21import math
22import numpy
23import os
24from acts import asserts
25from acts import base_test
26from acts import context
27from acts import utils
28from acts.controllers import iperf_server as ipf
29from acts.controllers.utils_lib import ssh
30from acts.metrics.loggers.blackbox import BlackboxMappedMetricLogger
31from acts.test_utils.wifi import ota_chamber
32from acts.test_utils.wifi import wifi_performance_test_utils as wputils
33from acts.test_utils.wifi import wifi_retail_ap as retail_ap
34from acts.test_utils.wifi import wifi_test_utils as wutils
35from functools import partial
36
37TEST_TIMEOUT = 10
38SHORT_SLEEP = 1
39MED_SLEEP = 6
40
41
42class WifiThroughputStabilityTest(base_test.BaseTestClass):
43    """Class to test WiFi throughput stability.
44
45    This class tests throughput stability and identifies cases where throughput
46    fluctuates over time. The class setups up the AP, configures and connects
47    the phone, and runs iperf throughput test at several attenuations For an
48    example config file to run this test class see
49    example_connectivity_performance_ap_sta.json.
50    """
51    def __init__(self, controllers):
52        base_test.BaseTestClass.__init__(self, controllers)
53        # Define metrics to be uploaded to BlackBox
54        self.testcase_metric_logger = (
55            BlackboxMappedMetricLogger.for_test_case())
56        self.testclass_metric_logger = (
57            BlackboxMappedMetricLogger.for_test_class())
58        self.publish_testcase_metrics = True
59        # Generate test cases
60        self.tests = self.generate_test_cases([6, 36, 149],
61                                              ['VHT20', 'VHT40', 'VHT80'],
62                                              ['TCP', 'UDP'], ['DL', 'UL'],
63                                              ['high', 'low'])
64
65    def generate_test_cases(self, channels, modes, traffic_types,
66                            traffic_directions, signal_levels):
67        """Function that auto-generates test cases for a test class."""
68        allowed_configs = {
69            'VHT20': [
70                1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 36, 40, 44, 48, 149, 153,
71                157, 161
72            ],
73            'VHT40': [36, 44, 149, 157],
74            'VHT80': [36, 149]
75        }
76        test_cases = []
77        for channel, mode, traffic_type, traffic_direction, signal_level in itertools.product(
78                channels, modes, traffic_types, traffic_directions,
79                signal_levels):
80            if channel not in allowed_configs[mode]:
81                continue
82            testcase_params = collections.OrderedDict(
83                channel=channel,
84                mode=mode,
85                traffic_type=traffic_type,
86                traffic_direction=traffic_direction,
87                signal_level=signal_level)
88            testcase_name = ('test_tput_stability'
89                             '_{}_{}_{}_ch{}_{}'.format(
90                                 signal_level, traffic_type, traffic_direction,
91                                 channel, mode))
92            setattr(self, testcase_name,
93                    partial(self._test_throughput_stability, testcase_params))
94            test_cases.append(testcase_name)
95        return test_cases
96
97    def setup_class(self):
98        self.dut = self.android_devices[0]
99        req_params = [
100            'throughput_stability_test_params', 'testbed_params',
101            'main_network', 'RetailAccessPoints', 'RemoteServer'
102        ]
103        opt_params = ['golden_files_list']
104        self.unpack_userparams(req_params, opt_params)
105        self.testclass_params = self.throughput_stability_test_params
106        self.num_atten = self.attenuators[0].instrument.num_atten
107        self.remote_server = ssh.connection.SshConnection(
108            ssh.settings.from_config(self.RemoteServer[0]['ssh_config']))
109        self.iperf_server = self.iperf_servers[0]
110        self.iperf_client = self.iperf_clients[0]
111        self.access_point = retail_ap.create(self.RetailAccessPoints)[0]
112        self.log_path = os.path.join(logging.log_path, 'test_results')
113        os.makedirs(self.log_path, exist_ok=True)
114        self.log.info('Access Point Configuration: {}'.format(
115            self.access_point.ap_settings))
116        if not hasattr(self, 'golden_files_list'):
117            self.golden_files_list = [
118                os.path.join(self.testbed_params['golden_results_path'], file)
119                for file in os.listdir(
120                    self.testbed_params['golden_results_path'])
121            ]
122        if hasattr(self, 'bdf'):
123            self.log.info('Pushing WiFi BDF to DUT.')
124            wputils.push_bdf(self.dut, self.bdf)
125        if hasattr(self, 'firmware'):
126            self.log.info('Pushing WiFi firmware to DUT.')
127            wlanmdsp = [
128                file for file in self.firmware if "wlanmdsp.mbn" in file
129            ][0]
130            data_msc = [file for file in self.firmware
131                        if "Data.msc" in file][0]
132            wputils.push_firmware(self.dut, wlanmdsp, data_msc)
133        self.testclass_results = []
134
135        # Turn WiFi ON
136        if self.testclass_params.get('airplane_mode', 1):
137            self.log.info('Turning on airplane mode.')
138            asserts.assert_true(utils.force_airplane_mode(self.dut, True),
139                                "Can not turn on airplane mode.")
140        wutils.wifi_toggle_state(self.dut, True)
141
142    def teardown_test(self):
143        self.iperf_server.stop()
144
145    def pass_fail_check(self, test_result_dict):
146        """Check the test result and decide if it passed or failed.
147
148        Checks the throughput stability test's PASS/FAIL criteria based on
149        minimum instantaneous throughput, and standard deviation.
150
151        Args:
152            test_result_dict: dict containing attenuation, throughput and other
153            meta data
154        """
155        avg_throughput = test_result_dict['iperf_results']['avg_throughput']
156        min_throughput = test_result_dict['iperf_results']['min_throughput']
157        std_dev_percent = (
158            test_result_dict['iperf_results']['std_deviation'] /
159            test_result_dict['iperf_results']['avg_throughput']) * 100
160        # Set blackbox metrics
161        if self.publish_testcase_metrics:
162            self.testcase_metric_logger.add_metric('avg_throughput',
163                                                   avg_throughput)
164            self.testcase_metric_logger.add_metric('min_throughput',
165                                                   min_throughput)
166            self.testcase_metric_logger.add_metric('std_dev_percent',
167                                                   std_dev_percent)
168        # Evaluate pass/fail
169        min_throughput_check = (
170            (min_throughput / avg_throughput) *
171            100) > self.testclass_params['min_throughput_threshold']
172        std_deviation_check = std_dev_percent < self.testclass_params[
173            'std_deviation_threshold']
174
175        if min_throughput_check and std_deviation_check:
176            asserts.explicit_pass(
177                'Test Passed. Throughput at {0:.2f}dB attenuation is stable. '
178                'Mean throughput is {1:.2f} Mbps with a standard deviation of '
179                '{2:.2f}% and dips down to {3:.2f} Mbps.'.format(
180                    test_result_dict['attenuation'], avg_throughput,
181                    std_dev_percent, min_throughput))
182        asserts.fail(
183            'Test Failed. Throughput at {0:.2f}dB attenuation is unstable. '
184            'Mean throughput is {1:.2f} Mbps with a standard deviation of '
185            '{2:.2f}% and dips down to {3:.2f} Mbps.'.format(
186                test_result_dict['attenuation'], avg_throughput,
187                std_dev_percent, min_throughput))
188
189    def post_process_results(self, test_result):
190        """Extracts results and saves plots and JSON formatted results.
191
192        Args:
193            test_result: dict containing attenuation, iPerfResult object and
194            other meta data
195        Returns:
196            test_result_dict: dict containing post-processed results including
197            avg throughput, other metrics, and other meta data
198        """
199        # Save output as text file
200        test_name = self.current_test_name
201        results_file_path = os.path.join(self.log_path,
202                                         '{}.txt'.format(test_name))
203        test_result_dict = {}
204        test_result_dict['ap_settings'] = test_result['ap_settings'].copy()
205        test_result_dict['attenuation'] = test_result['attenuation']
206        if test_result['iperf_result'].instantaneous_rates:
207            instantaneous_rates_Mbps = [
208                rate * 8 * (1.024**2)
209                for rate in test_result['iperf_result'].instantaneous_rates[
210                    self.testclass_params['iperf_ignored_interval']:-1]
211            ]
212        else:
213            instantaneous_rates_Mbps = float('nan')
214        test_result_dict['iperf_results'] = {
215            'instantaneous_rates':
216            instantaneous_rates_Mbps,
217            'avg_throughput':
218            numpy.mean(instantaneous_rates_Mbps),
219            'std_deviation':
220            test_result['iperf_result'].get_std_deviation(
221                self.testclass_params['iperf_ignored_interval']) * 8,
222            'min_throughput':
223            min(instantaneous_rates_Mbps)
224        }
225        with open(results_file_path, 'w') as results_file:
226            json.dump(test_result_dict, results_file)
227        # Plot and save
228        figure = wputils.BokehFigure(test_name,
229                                     x_label='Time (s)',
230                                     primary_y_label='Throughput (Mbps)')
231        time_data = list(range(0, len(instantaneous_rates_Mbps)))
232        figure.add_line(time_data,
233                        instantaneous_rates_Mbps,
234                        legend=self.current_test_name,
235                        marker='circle')
236        output_file_path = os.path.join(self.log_path,
237                                        '{}.html'.format(test_name))
238        figure.generate_figure(output_file_path)
239        return test_result_dict
240
241    def setup_ap(self, testcase_params):
242        """Sets up the access point in the configuration required by the test.
243
244        Args:
245            testcase_params: dict containing AP and other test params
246        """
247        band = self.access_point.band_lookup_by_channel(
248            testcase_params['channel'])
249        if '2G' in band:
250            frequency = wutils.WifiEnums.channel_2G_to_freq[
251                testcase_params['channel']]
252        else:
253            frequency = wutils.WifiEnums.channel_5G_to_freq[
254                testcase_params['channel']]
255        if frequency in wutils.WifiEnums.DFS_5G_FREQUENCIES:
256            self.access_point.set_region(self.testbed_params['DFS_region'])
257        else:
258            self.access_point.set_region(self.testbed_params['default_region'])
259        self.access_point.set_channel(band, testcase_params['channel'])
260        self.access_point.set_bandwidth(band, testcase_params['mode'])
261        self.log.info('Access Point Configuration: {}'.format(
262            self.access_point.ap_settings))
263
264    def setup_dut(self, testcase_params):
265        """Sets up the DUT in the configuration required by the test.
266
267        Args:
268            testcase_params: dict containing AP and other test params
269        """
270        # Check battery level before test
271        if not wputils.health_check(self.dut, 10):
272            asserts.skip('Battery level too low. Skipping test.')
273        # Turn screen off to preserve battery
274        self.dut.go_to_sleep()
275        band = self.access_point.band_lookup_by_channel(
276            testcase_params['channel'])
277        if wputils.validate_network(self.dut,
278                                    testcase_params['test_network']['SSID']):
279            self.log.info('Already connected to desired network')
280        else:
281            wutils.wifi_toggle_state(self.dut, True)
282            wutils.reset_wifi(self.dut)
283            wutils.set_wifi_country_code(self.dut,
284                self.testclass_params['country_code'])
285            self.main_network[band]['channel'] = testcase_params['channel']
286            wutils.wifi_connect(self.dut,
287                                testcase_params['test_network'],
288                                num_of_tries=5,
289                                check_connectivity=False)
290        self.dut_ip = self.dut.droid.connectivityGetIPv4Addresses('wlan0')[0]
291
292    def setup_throughput_stability_test(self, testcase_params):
293        """Function that gets devices ready for the test.
294
295        Args:
296            testcase_params: dict containing test-specific parameters
297        """
298        # Configure AP
299        self.setup_ap(testcase_params)
300        # Set attenuator to 0 dB
301        self.log.info('Setting attenuation to {} dB'.format(
302            testcase_params['atten_level']))
303        for attenuator in self.attenuators:
304            attenuator.set_atten(testcase_params['atten_level'])
305        # Reset, configure, and connect DUT
306        self.setup_dut(testcase_params)
307        if isinstance(self.iperf_server, ipf.IPerfServerOverAdb):
308            testcase_params['iperf_server_address'] = self.dut_ip
309        else:
310            testcase_params[
311                'iperf_server_address'] = wputils.get_server_address(
312                    self.remote_server, self.dut_ip, '255.255.255.0')
313
314    def run_throughput_stability_test(self, testcase_params):
315        """Main function to test throughput stability.
316
317        The function sets up the AP in the correct channel and mode
318        configuration and runs an iperf test to measure throughput.
319
320        Args:
321            testcase_params: dict containing test specific parameters
322        Returns:
323            test_result: dict containing test result and meta data
324        """
325        # Run test and log result
326        # Start iperf session
327        self.log.info('Starting iperf test.')
328        self.iperf_server.start(tag=str(testcase_params['atten_level']))
329        client_output_path = self.iperf_client.start(
330            testcase_params['iperf_server_address'],
331            testcase_params['iperf_args'], str(testcase_params['atten_level']),
332            self.testclass_params['iperf_duration'] + TEST_TIMEOUT)
333        server_output_path = self.iperf_server.stop()
334        # Set attenuator to 0 dB
335        for attenuator in self.attenuators:
336            attenuator.set_atten(0)
337        # Parse and log result
338        if testcase_params['use_client_output']:
339            iperf_file = client_output_path
340        else:
341            iperf_file = server_output_path
342        try:
343            iperf_result = ipf.IPerfResult(iperf_file)
344        except:
345            asserts.fail('Cannot get iperf result.')
346        test_result = collections.OrderedDict()
347        test_result['testcase_params'] = testcase_params.copy()
348        test_result['ap_settings'] = self.access_point.ap_settings.copy()
349        test_result['attenuation'] = testcase_params['atten_level']
350        test_result['iperf_result'] = iperf_result
351        self.testclass_results.append(test_result)
352        return test_result
353
354    def get_target_atten_tput(self, testcase_params):
355        """Function gets attenuation used for test
356
357        The function fetches the attenuation at which the test should be
358        performed, and the expected target average throughput.
359
360        Args:
361            testcase_params: dict containing test specific parameters
362        Returns:
363            test_target: dict containing target test attenuation and expected
364            throughput
365        """
366        # Fetch the golden RvR results
367        rvr_golden_file_name = 'test_rvr_' + '_'.join(
368            self.current_test_name.split('_')[4:])
369        try:
370            golden_path = next(file_name
371                               for file_name in self.golden_files_list
372                               if rvr_golden_file_name in file_name)
373        except:
374            asserts.fail('Test failed. Golden data not found.')
375
376        with open(golden_path, 'r') as golden_file:
377            golden_results = json.load(golden_file)
378        test_target = {}
379        if testcase_params['signal_level'] == 'low':
380            # Get last test point where throughput is above
381            throughput_below_target = [
382                x < self.testclass_params['low_throughput_target']
383                for x in golden_results['throughput_receive']
384            ]
385            atten_idx = throughput_below_target.index(1) - 1
386            test_target['target_attenuation'] = golden_results['attenuation'][
387                atten_idx]
388            test_target['target_throughput'] = golden_results[
389                'throughput_receive'][atten_idx]
390        if testcase_params['signal_level'] == 'high':
391            # Test at lowest attenuation point
392            test_target['target_attenuation'] = golden_results['attenuation'][
393                0]
394            test_target['target_throughput'] = golden_results[
395                'throughput_receive'][0]
396        return test_target
397
398    def compile_test_params(self, testcase_params):
399        """Function that completes setting the test case parameters."""
400        band = self.access_point.band_lookup_by_channel(
401            testcase_params['channel'])
402        testcase_params['test_network'] = self.main_network[band]
403        testcase_params['test_target'] = self.get_target_atten_tput(
404            testcase_params)
405        testcase_params['atten_level'] = testcase_params['test_target'][
406            'target_attenuation']
407        self.atten_level = testcase_params['atten_level']
408
409        if (testcase_params['traffic_direction'] == 'DL'
410                and not isinstance(self.iperf_server, ipf.IPerfServerOverAdb)
411            ) or (testcase_params['traffic_direction'] == 'UL'
412                  and isinstance(self.iperf_server, ipf.IPerfServerOverAdb)):
413            testcase_params['iperf_args'] = wputils.get_iperf_arg_string(
414                duration=self.testclass_params['iperf_duration'],
415                reverse_direction=1,
416                traffic_type=testcase_params['traffic_type'])
417            testcase_params['use_client_output'] = True
418        else:
419            testcase_params['iperf_args'] = wputils.get_iperf_arg_string(
420                duration=self.testclass_params['iperf_duration'],
421                reverse_direction=0,
422                traffic_type=testcase_params['traffic_type'])
423            testcase_params['use_client_output'] = False
424
425        return testcase_params
426
427    def _test_throughput_stability(self, testcase_params):
428        """ Function that gets called for each test case
429
430        The function gets called in each test case. The function customizes
431        the test based on the test name of the test that called it
432
433        Args:
434            testcase_params: dict containing test specific parameters
435        """
436        testcase_params = self.compile_test_params(testcase_params)
437        self.setup_throughput_stability_test(testcase_params)
438        test_result = self.run_throughput_stability_test(testcase_params)
439        test_result_postprocessed = self.post_process_results(test_result)
440        self.pass_fail_check(test_result_postprocessed)
441
442
443# Over-the air version of ping tests
444class WifiOtaThroughputStabilityTest(WifiThroughputStabilityTest):
445    """Class to test over-the-air ping
446
447    This class tests WiFi ping performance in an OTA chamber. It enables
448    setting turntable orientation and other chamber parameters to study
449    performance in varying channel conditions
450    """
451    def __init__(self, controllers):
452        base_test.BaseTestClass.__init__(self, controllers)
453        # Define metrics to be uploaded to BlackBox
454        self.testcase_metric_logger = (
455            BlackboxMappedMetricLogger.for_test_case())
456        self.testclass_metric_logger = (
457            BlackboxMappedMetricLogger.for_test_class())
458        self.publish_testcase_metrics = False
459
460    def setup_class(self):
461        WifiThroughputStabilityTest.setup_class(self)
462        self.ota_chamber = ota_chamber.create(
463            self.user_params['OTAChamber'])[0]
464
465    def teardown_class(self):
466        self.ota_chamber.reset_chamber()
467        self.process_testclass_results()
468
469    def extract_test_id(self, testcase_params, id_fields):
470        test_id = collections.OrderedDict(
471            (param, testcase_params[param]) for param in id_fields)
472        return test_id
473
474    def process_testclass_results(self):
475        """Saves all test results to enable comparison."""
476        testclass_data = collections.OrderedDict()
477        for test in self.testclass_results:
478            current_params = test['testcase_params']
479            channel_data = testclass_data.setdefault(current_params['channel'],
480                                                     collections.OrderedDict())
481            test_id = tuple(
482                self.extract_test_id(current_params, [
483                    'mode', 'traffic_type', 'traffic_direction', 'signal_level'
484                ]).items())
485            test_data = channel_data.setdefault(
486                test_id, collections.OrderedDict(position=[], throughput=[]))
487            current_throughput = (numpy.mean(
488                test['iperf_result'].instantaneous_rates[
489                    self.testclass_params['iperf_ignored_interval']:-1])
490                                  ) * 8 * (1.024**2)
491            test_data['position'].append(current_params['position'])
492            test_data['throughput'].append(current_throughput)
493
494        chamber_mode = self.testclass_results[0]['testcase_params'][
495            'chamber_mode']
496        if chamber_mode == 'orientation':
497            x_label = 'Angle (deg)'
498        elif chamber_mode == 'stepped stirrers':
499            x_label = 'Position Index'
500
501        # Publish test class metrics
502        for channel, channel_data in testclass_data.items():
503            for test_id, test_data in channel_data.items():
504                test_id_dict = dict(test_id)
505                metric_tag = 'ota_summary_{}_{}_{}_ch{}_{}'.format(
506                    test_id_dict['signal_level'], test_id_dict['traffic_type'],
507                    test_id_dict['traffic_direction'], channel,
508                    test_id_dict['mode'])
509                metric_name = metric_tag + '.avg_throughput'
510                metric_value = numpy.mean(test_data['throughput'])
511                self.testclass_metric_logger.add_metric(
512                    metric_name, metric_value)
513                metric_name = metric_tag + '.min_throughput'
514                metric_value = min(test_data['throughput'])
515                self.testclass_metric_logger.add_metric(
516                    metric_name, metric_value)
517
518        # Plot test class results
519        plots = []
520        for channel, channel_data in testclass_data.items():
521            current_plot = wputils.BokehFigure(
522                title='Channel {} - Rate vs. Position'.format(channel),
523                x_label=x_label,
524                primary_y_label='Rate (Mbps)',
525            )
526            for test_id, test_data in channel_data.items():
527                test_id_dict = dict(test_id)
528                legend = '{}, {} {}, {} RSSI'.format(
529                    test_id_dict['mode'], test_id_dict['traffic_type'],
530                    test_id_dict['traffic_direction'],
531                    test_id_dict['signal_level'])
532                current_plot.add_line(test_data['position'],
533                                      test_data['throughput'], legend)
534            current_plot.generate_figure()
535            plots.append(current_plot)
536        current_context = context.get_current_context().get_full_output_path()
537        plot_file_path = os.path.join(current_context, 'results.html')
538        wputils.BokehFigure.save_figures(plots, plot_file_path)
539
540    def setup_throughput_stability_test(self, testcase_params):
541        WifiThroughputStabilityTest.setup_throughput_stability_test(
542            self, testcase_params)
543        # Setup turntable
544        if testcase_params['chamber_mode'] == 'orientation':
545            self.ota_chamber.set_orientation(testcase_params['position'])
546        elif testcase_params['chamber_mode'] == 'stepped stirrers':
547            self.ota_chamber.step_stirrers(testcase_params['total_positions'])
548
549    def get_target_atten_tput(self, testcase_params):
550        test_target = {}
551        if testcase_params['signal_level'] == 'high':
552            test_target['target_attenuation'] = self.testclass_params[
553                'default_atten_levels'][0]
554        elif testcase_params['signal_level'] == 'low':
555            test_target['target_attenuation'] = self.testclass_params[
556                'default_atten_levels'][1]
557        test_target['target_throughput'] = 0
558        return test_target
559
560    def generate_test_cases(self, channels, modes, traffic_types,
561                            traffic_directions, signal_levels, chamber_mode,
562                            positions):
563        allowed_configs = {
564            'VHT20': [
565                1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 36, 40, 44, 48, 149, 153,
566                157, 161
567            ],
568            'VHT40': [36, 44, 149, 157],
569            'VHT80': [36, 149]
570        }
571        test_cases = []
572        for channel, mode, position, traffic_type, signal_level, traffic_direction in itertools.product(
573                channels, modes, positions, traffic_types, signal_levels,
574                traffic_directions):
575            if channel not in allowed_configs[mode]:
576                continue
577            testcase_params = collections.OrderedDict(
578                channel=channel,
579                mode=mode,
580                traffic_type=traffic_type,
581                traffic_direction=traffic_direction,
582                signal_level=signal_level,
583                chamber_mode=chamber_mode,
584                total_positions=len(positions),
585                position=position)
586            testcase_name = ('test_tput_stability'
587                             '_{}_{}_{}_ch{}_{}_pos{}'.format(
588                                 signal_level, traffic_type, traffic_direction,
589                                 channel, mode, position))
590            setattr(self, testcase_name,
591                    partial(self._test_throughput_stability, testcase_params))
592            test_cases.append(testcase_name)
593        return test_cases
594
595
596class WifiOtaThroughputStability_TenDegree_Test(WifiOtaThroughputStabilityTest
597                                                ):
598    def __init__(self, controllers):
599        WifiOtaThroughputStabilityTest.__init__(self, controllers)
600        self.tests = self.generate_test_cases([6, 36, 149], ['VHT20', 'VHT80'],
601                                              ['TCP'], ['DL', 'UL'],
602                                              ['high', 'low'], 'orientation',
603                                              list(range(0, 360, 10)))
604
605
606class WifiOtaThroughputStability_45Degree_Test(WifiOtaThroughputStabilityTest):
607    def __init__(self, controllers):
608        WifiOtaThroughputStabilityTest.__init__(self, controllers)
609        self.tests = self.generate_test_cases([6, 36, 149], ['VHT20', 'VHT80'],
610                                              ['TCP'], ['DL', 'UL'],
611                                              ['high', 'low'], 'orientation',
612                                              list(range(0, 360, 45)))
613
614
615class WifiOtaThroughputStability_SteppedStirrers_Test(
616        WifiOtaThroughputStabilityTest):
617    def __init__(self, controllers):
618        WifiOtaThroughputStabilityTest.__init__(self, controllers)
619        self.tests = self.generate_test_cases([6, 36, 149], ['VHT20', 'VHT80'],
620                                              ['TCP'], ['DL', 'UL'],
621                                              ['high', 'low'],
622                                              'stepped stirrers',
623                                              list(range(100)))
624