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 numpy
22import os
23import time
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, signal_level, traffic_type, traffic_direction in itertools.product(
78                channels,
79                modes,
80                signal_levels,
81                traffic_types,
82                traffic_directions,
83        ):
84            if channel not in allowed_configs[mode]:
85                continue
86            testcase_params = collections.OrderedDict(
87                channel=channel,
88                mode=mode,
89                traffic_type=traffic_type,
90                traffic_direction=traffic_direction,
91                signal_level=signal_level)
92            testcase_name = ('test_tput_stability'
93                             '_{}_{}_{}_ch{}_{}'.format(
94                                 signal_level, traffic_type, traffic_direction,
95                                 channel, mode))
96            setattr(self, testcase_name,
97                    partial(self._test_throughput_stability, testcase_params))
98            test_cases.append(testcase_name)
99        return test_cases
100
101    def setup_class(self):
102        self.dut = self.android_devices[0]
103        req_params = [
104            'throughput_stability_test_params', 'testbed_params',
105            'main_network', 'RetailAccessPoints', 'RemoteServer'
106        ]
107        self.unpack_userparams(req_params)
108        self.testclass_params = self.throughput_stability_test_params
109        self.num_atten = self.attenuators[0].instrument.num_atten
110        self.remote_server = ssh.connection.SshConnection(
111            ssh.settings.from_config(self.RemoteServer[0]['ssh_config']))
112        self.iperf_server = self.iperf_servers[0]
113        self.iperf_client = self.iperf_clients[0]
114        self.access_point = retail_ap.create(self.RetailAccessPoints)[0]
115        self.log_path = os.path.join(logging.log_path, 'test_results')
116        os.makedirs(self.log_path, exist_ok=True)
117        self.log.info('Access Point Configuration: {}'.format(
118            self.access_point.ap_settings))
119        self.ref_attenuations = {}
120        self.testclass_results = []
121
122        # Turn WiFi ON
123        if self.testclass_params.get('airplane_mode', 1):
124            self.log.info('Turning on airplane mode.')
125            asserts.assert_true(utils.force_airplane_mode(self.dut, True),
126                                "Can not turn on airplane mode.")
127        wutils.wifi_toggle_state(self.dut, True)
128
129    def teardown_test(self):
130        self.iperf_server.stop()
131
132    def pass_fail_check(self, test_result_dict):
133        """Check the test result and decide if it passed or failed.
134
135        Checks the throughput stability test's PASS/FAIL criteria based on
136        minimum instantaneous throughput, and standard deviation.
137
138        Args:
139            test_result_dict: dict containing attenuation, throughput and other
140            meta data
141        """
142        avg_throughput = test_result_dict['iperf_results']['avg_throughput']
143        min_throughput = test_result_dict['iperf_results']['min_throughput']
144        std_dev_percent = (
145            test_result_dict['iperf_results']['std_deviation'] /
146            test_result_dict['iperf_results']['avg_throughput']) * 100
147        # Set blackbox metrics
148        if self.publish_testcase_metrics:
149            self.testcase_metric_logger.add_metric('avg_throughput',
150                                                   avg_throughput)
151            self.testcase_metric_logger.add_metric('min_throughput',
152                                                   min_throughput)
153            self.testcase_metric_logger.add_metric('std_dev_percent',
154                                                   std_dev_percent)
155        # Evaluate pass/fail
156        min_throughput_check = (
157            (min_throughput / avg_throughput) *
158            100) > self.testclass_params['min_throughput_threshold']
159        std_deviation_check = std_dev_percent < self.testclass_params[
160            'std_deviation_threshold']
161
162        test_message = (
163            'Atten: {0:.2f}dB, RSSI: {1:.2f}dB. '
164            'Throughput (Mean: {2:.2f}, Std. Dev:{3:.2f}%, Min: {4:.2f} Mbps).'
165            'LLStats : {5}'.format(test_result_dict['attenuation'],
166                                   test_result_dict['rssi'], avg_throughput,
167                                   std_dev_percent, min_throughput,
168                                   test_result_dict['llstats']))
169        if min_throughput_check and std_deviation_check:
170            asserts.explicit_pass('Test Passed.' + test_message)
171        asserts.fail('Test Failed. ' + test_message)
172
173    def post_process_results(self, test_result):
174        """Extracts results and saves plots and JSON formatted results.
175
176        Args:
177            test_result: dict containing attenuation, iPerfResult object and
178            other meta data
179        Returns:
180            test_result_dict: dict containing post-processed results including
181            avg throughput, other metrics, and other meta data
182        """
183        # Save output as text file
184        test_name = self.current_test_name
185        results_file_path = os.path.join(self.log_path,
186                                         '{}.txt'.format(test_name))
187        test_result_dict = {}
188        test_result_dict['ap_settings'] = test_result['ap_settings'].copy()
189        test_result_dict['attenuation'] = test_result['attenuation']
190        test_result_dict['rssi'] = test_result['rssi_result'][
191            'signal_poll_rssi']['mean']
192        test_result_dict['llstats'] = (
193            'TX MCS = {0} ({1:.1f}%). '
194            'RX MCS = {2} ({3:.1f}%)'.format(
195                test_result['llstats']['summary']['common_tx_mcs'],
196                test_result['llstats']['summary']['common_tx_mcs_freq'] * 100,
197                test_result['llstats']['summary']['common_rx_mcs'],
198                test_result['llstats']['summary']['common_rx_mcs_freq'] * 100))
199        if test_result['iperf_result'].instantaneous_rates:
200            instantaneous_rates_Mbps = [
201                rate * 8 * (1.024**2)
202                for rate in test_result['iperf_result'].instantaneous_rates[
203                    self.testclass_params['iperf_ignored_interval']:-1]
204            ]
205            tput_standard_deviation = test_result[
206                'iperf_result'].get_std_deviation(
207                    self.testclass_params['iperf_ignored_interval']) * 8
208        else:
209            instantaneous_rates_Mbps = float('nan')
210            tput_standard_deviation = float('nan')
211        test_result_dict['iperf_results'] = {
212            'instantaneous_rates': instantaneous_rates_Mbps,
213            'avg_throughput': numpy.mean(instantaneous_rates_Mbps),
214            'std_deviation': tput_standard_deviation,
215            'min_throughput': min(instantaneous_rates_Mbps)
216        }
217        with open(results_file_path, 'w') as results_file:
218            json.dump(test_result_dict, results_file)
219        # Plot and save
220        figure = wputils.BokehFigure(test_name,
221                                     x_label='Time (s)',
222                                     primary_y_label='Throughput (Mbps)')
223        time_data = list(range(0, len(instantaneous_rates_Mbps)))
224        figure.add_line(time_data,
225                        instantaneous_rates_Mbps,
226                        legend=self.current_test_name,
227                        marker='circle')
228        output_file_path = os.path.join(self.log_path,
229                                        '{}.html'.format(test_name))
230        figure.generate_figure(output_file_path)
231        return test_result_dict
232
233    def setup_ap(self, testcase_params):
234        """Sets up the access point in the configuration required by the test.
235
236        Args:
237            testcase_params: dict containing AP and other test params
238        """
239        band = self.access_point.band_lookup_by_channel(
240            testcase_params['channel'])
241        if '2G' in band:
242            frequency = wutils.WifiEnums.channel_2G_to_freq[
243                testcase_params['channel']]
244        else:
245            frequency = wutils.WifiEnums.channel_5G_to_freq[
246                testcase_params['channel']]
247        if frequency in wutils.WifiEnums.DFS_5G_FREQUENCIES:
248            self.access_point.set_region(self.testbed_params['DFS_region'])
249        else:
250            self.access_point.set_region(self.testbed_params['default_region'])
251        self.access_point.set_channel(band, testcase_params['channel'])
252        self.access_point.set_bandwidth(band, testcase_params['mode'])
253        self.log.info('Access Point Configuration: {}'.format(
254            self.access_point.ap_settings))
255
256    def setup_dut(self, testcase_params):
257        """Sets up the DUT in the configuration required by the test.
258
259        Args:
260            testcase_params: dict containing AP and other test params
261        """
262        # Check battery level before test
263        if not wputils.health_check(self.dut, 10):
264            asserts.skip('Battery level too low. Skipping test.')
265        # Turn screen off to preserve battery
266        self.dut.go_to_sleep()
267        band = self.access_point.band_lookup_by_channel(
268            testcase_params['channel'])
269        if wputils.validate_network(self.dut,
270                                    testcase_params['test_network']['SSID']):
271            self.log.info('Already connected to desired network')
272        else:
273            wutils.wifi_toggle_state(self.dut, True)
274            wutils.reset_wifi(self.dut)
275            wutils.set_wifi_country_code(self.dut,
276                                         self.testclass_params['country_code'])
277            self.main_network[band]['channel'] = testcase_params['channel']
278            wutils.wifi_connect(self.dut,
279                                testcase_params['test_network'],
280                                num_of_tries=5,
281                                check_connectivity=False)
282        self.dut_ip = self.dut.droid.connectivityGetIPv4Addresses('wlan0')[0]
283
284    def setup_throughput_stability_test(self, testcase_params):
285        """Function that gets devices ready for the test.
286
287        Args:
288            testcase_params: dict containing test-specific parameters
289        """
290        # Configure AP
291        self.setup_ap(testcase_params)
292        # Reset, configure, and connect DUT
293        self.setup_dut(testcase_params)
294        # Wait before running the first wifi test
295        first_test_delay = self.testclass_params.get('first_test_delay', 600)
296        if first_test_delay > 0 and len(self.testclass_results) == 0:
297            self.log.info('Waiting before the first test.')
298            time.sleep(first_test_delay)
299            self.setup_dut(testcase_params)
300        # Get and set attenuation levels for test
301        testcase_params['atten_level'] = self.get_target_atten(testcase_params)
302        self.log.info('Setting attenuation to {} dB'.format(
303            testcase_params['atten_level']))
304        for attenuator in self.attenuators:
305            attenuator.set_atten(testcase_params['atten_level'])
306        # Configure iperf
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        llstats_obj = wputils.LinkLayerStats(self.dut)
329        llstats_obj.update_stats()
330        self.iperf_server.start(tag=str(testcase_params['atten_level']))
331        current_rssi = wputils.get_connected_rssi_nb(
332            dut=self.dut,
333            num_measurements=self.testclass_params['iperf_duration'] - 1,
334            polling_frequency=1,
335            first_measurement_delay=1,
336            disconnect_warning=1,
337            ignore_samples=1)
338        client_output_path = self.iperf_client.start(
339            testcase_params['iperf_server_address'],
340            testcase_params['iperf_args'], str(testcase_params['atten_level']),
341            self.testclass_params['iperf_duration'] + TEST_TIMEOUT)
342        current_rssi = current_rssi.result()
343        server_output_path = self.iperf_server.stop()
344        # Set attenuator to 0 dB
345        for attenuator in self.attenuators:
346            attenuator.set_atten(0)
347        # Parse and log result
348        if testcase_params['use_client_output']:
349            iperf_file = client_output_path
350        else:
351            iperf_file = server_output_path
352        try:
353            iperf_result = ipf.IPerfResult(iperf_file)
354        except:
355            asserts.fail('Cannot get iperf result.')
356        llstats_obj.update_stats()
357        curr_llstats = llstats_obj.llstats_incremental.copy()
358        test_result = collections.OrderedDict()
359        test_result['testcase_params'] = testcase_params.copy()
360        test_result['ap_settings'] = self.access_point.ap_settings.copy()
361        test_result['attenuation'] = testcase_params['atten_level']
362        test_result['iperf_result'] = iperf_result
363        test_result['rssi_result'] = current_rssi
364        test_result['llstats'] = curr_llstats
365        self.testclass_results.append(test_result)
366        return test_result
367
368    def get_target_atten(self, testcase_params):
369        """Function gets attenuation used for test
370
371        The function fetches the attenuation at which the test should be
372        performed.
373
374        Args:
375            testcase_params: dict containing test specific parameters
376        Returns:
377            test_atten: target attenuation for test
378        """
379        # Get attenuation from reference test if it has been run
380        ref_test_fields = ['channel', 'mode', 'signal_level']
381        test_id = wputils.extract_sub_dict(testcase_params, ref_test_fields)
382        test_id = tuple(test_id.items())
383        if test_id in self.ref_attenuations:
384            return self.ref_attenuations[test_id]
385
386        # Get attenuation for target RSSI
387        if testcase_params['signal_level'] == 'low':
388            target_rssi = self.testclass_params['low_throughput_target']
389        else:
390            target_rssi = self.testclass_params['high_throughput_target']
391        target_atten = wputils.get_atten_for_target_rssi(
392            target_rssi, self.attenuators, self.dut, self.remote_server)
393
394        self.ref_attenuations[test_id] = target_atten
395        return self.ref_attenuations[test_id]
396
397    def compile_test_params(self, testcase_params):
398        """Function that completes setting the test case parameters."""
399        band = self.access_point.band_lookup_by_channel(
400            testcase_params['channel'])
401        testcase_params['test_network'] = self.main_network[band]
402
403        if (testcase_params['traffic_direction'] == 'DL'
404                and not isinstance(self.iperf_server, ipf.IPerfServerOverAdb)
405            ) or (testcase_params['traffic_direction'] == 'UL'
406                  and isinstance(self.iperf_server, ipf.IPerfServerOverAdb)):
407            testcase_params['iperf_args'] = wputils.get_iperf_arg_string(
408                duration=self.testclass_params['iperf_duration'],
409                reverse_direction=1,
410                traffic_type=testcase_params['traffic_type'])
411            testcase_params['use_client_output'] = True
412        else:
413            testcase_params['iperf_args'] = wputils.get_iperf_arg_string(
414                duration=self.testclass_params['iperf_duration'],
415                reverse_direction=0,
416                traffic_type=testcase_params['traffic_type'])
417            testcase_params['use_client_output'] = False
418
419        return testcase_params
420
421    def _test_throughput_stability(self, testcase_params):
422        """ Function that gets called for each test case
423
424        The function gets called in each test case. The function customizes
425        the test based on the test name of the test that called it
426
427        Args:
428            testcase_params: dict containing test specific parameters
429        """
430        testcase_params = self.compile_test_params(testcase_params)
431        self.setup_throughput_stability_test(testcase_params)
432        test_result = self.run_throughput_stability_test(testcase_params)
433        test_result_postprocessed = self.post_process_results(test_result)
434        self.pass_fail_check(test_result_postprocessed)
435
436
437# Over-the air version of ping tests
438class WifiOtaThroughputStabilityTest(WifiThroughputStabilityTest):
439    """Class to test over-the-air ping
440
441    This class tests WiFi ping performance in an OTA chamber. It enables
442    setting turntable orientation and other chamber parameters to study
443    performance in varying channel conditions
444    """
445    def __init__(self, controllers):
446        base_test.BaseTestClass.__init__(self, controllers)
447        # Define metrics to be uploaded to BlackBox
448        self.testcase_metric_logger = (
449            BlackboxMappedMetricLogger.for_test_case())
450        self.testclass_metric_logger = (
451            BlackboxMappedMetricLogger.for_test_class())
452        self.publish_testcase_metrics = False
453
454    def setup_class(self):
455        WifiThroughputStabilityTest.setup_class(self)
456        self.ota_chamber = ota_chamber.create(
457            self.user_params['OTAChamber'])[0]
458
459    def teardown_class(self):
460        self.ota_chamber.reset_chamber()
461        self.process_testclass_results()
462
463    def extract_test_id(self, testcase_params, id_fields):
464        test_id = collections.OrderedDict(
465            (param, testcase_params[param]) for param in id_fields)
466        return test_id
467
468    def process_testclass_results(self):
469        """Saves all test results to enable comparison."""
470        testclass_data = collections.OrderedDict()
471        for test in self.testclass_results:
472            current_params = test['testcase_params']
473            channel_data = testclass_data.setdefault(current_params['channel'],
474                                                     collections.OrderedDict())
475            test_id = tuple(
476                self.extract_test_id(current_params, [
477                    'mode', 'traffic_type', 'traffic_direction', 'signal_level'
478                ]).items())
479            test_data = channel_data.setdefault(
480                test_id, collections.OrderedDict(position=[], throughput=[]))
481            current_throughput = (numpy.mean(
482                test['iperf_result'].instantaneous_rates[
483                    self.testclass_params['iperf_ignored_interval']:-1])
484                                  ) * 8 * (1.024**2)
485            test_data['position'].append(current_params['position'])
486            test_data['throughput'].append(current_throughput)
487
488        chamber_mode = self.testclass_results[0]['testcase_params'][
489            'chamber_mode']
490        if chamber_mode == 'orientation':
491            x_label = 'Angle (deg)'
492        elif chamber_mode == 'stepped stirrers':
493            x_label = 'Position Index'
494
495        # Publish test class metrics
496        for channel, channel_data in testclass_data.items():
497            for test_id, test_data in channel_data.items():
498                test_id_dict = dict(test_id)
499                metric_tag = 'ota_summary_{}_{}_{}_ch{}_{}'.format(
500                    test_id_dict['signal_level'], test_id_dict['traffic_type'],
501                    test_id_dict['traffic_direction'], channel,
502                    test_id_dict['mode'])
503                metric_name = metric_tag + '.avg_throughput'
504                metric_value = numpy.mean(test_data['throughput'])
505                self.testclass_metric_logger.add_metric(
506                    metric_name, metric_value)
507                metric_name = metric_tag + '.min_throughput'
508                metric_value = min(test_data['throughput'])
509                self.testclass_metric_logger.add_metric(
510                    metric_name, metric_value)
511
512        # Plot test class results
513        plots = []
514        for channel, channel_data in testclass_data.items():
515            current_plot = wputils.BokehFigure(
516                title='Channel {} - Rate vs. Position'.format(channel),
517                x_label=x_label,
518                primary_y_label='Rate (Mbps)',
519            )
520            for test_id, test_data in channel_data.items():
521                test_id_dict = dict(test_id)
522                legend = '{}, {} {}, {} RSSI'.format(
523                    test_id_dict['mode'], test_id_dict['traffic_type'],
524                    test_id_dict['traffic_direction'],
525                    test_id_dict['signal_level'])
526                current_plot.add_line(test_data['position'],
527                                      test_data['throughput'], legend)
528            current_plot.generate_figure()
529            plots.append(current_plot)
530        current_context = context.get_current_context().get_full_output_path()
531        plot_file_path = os.path.join(current_context, 'results.html')
532        wputils.BokehFigure.save_figures(plots, plot_file_path)
533
534    def setup_throughput_stability_test(self, testcase_params):
535        WifiThroughputStabilityTest.setup_throughput_stability_test(
536            self, testcase_params)
537        # Setup turntable
538        if testcase_params['chamber_mode'] == 'orientation':
539            self.ota_chamber.set_orientation(testcase_params['position'])
540        elif testcase_params['chamber_mode'] == 'stepped stirrers':
541            self.ota_chamber.step_stirrers(testcase_params['total_positions'])
542
543    def get_target_atten(self, testcase_params):
544        if testcase_params['signal_level'] == 'high':
545            test_atten = self.testclass_params['default_atten_levels'][0]
546        elif testcase_params['signal_level'] == 'low':
547            test_atten = self.testclass_params['default_atten_levels'][1]
548        return test_atten
549
550    def generate_test_cases(self, channels, modes, traffic_types,
551                            traffic_directions, signal_levels, chamber_mode,
552                            positions):
553        allowed_configs = {
554            'VHT20': [
555                1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 36, 40, 44, 48, 149, 153,
556                157, 161
557            ],
558            'VHT40': [36, 44, 149, 157],
559            'VHT80': [36, 149]
560        }
561        test_cases = []
562        for channel, mode, position, traffic_type, signal_level, traffic_direction in itertools.product(
563                channels, modes, positions, traffic_types, signal_levels,
564                traffic_directions):
565            if channel not in allowed_configs[mode]:
566                continue
567            testcase_params = collections.OrderedDict(
568                channel=channel,
569                mode=mode,
570                traffic_type=traffic_type,
571                traffic_direction=traffic_direction,
572                signal_level=signal_level,
573                chamber_mode=chamber_mode,
574                total_positions=len(positions),
575                position=position)
576            testcase_name = ('test_tput_stability'
577                             '_{}_{}_{}_ch{}_{}_pos{}'.format(
578                                 signal_level, traffic_type, traffic_direction,
579                                 channel, mode, position))
580            setattr(self, testcase_name,
581                    partial(self._test_throughput_stability, testcase_params))
582            test_cases.append(testcase_name)
583        return test_cases
584
585
586class WifiOtaThroughputStability_TenDegree_Test(WifiOtaThroughputStabilityTest
587                                                ):
588    def __init__(self, controllers):
589        WifiOtaThroughputStabilityTest.__init__(self, controllers)
590        self.tests = self.generate_test_cases([6, 36, 149], ['VHT20', 'VHT80'],
591                                              ['TCP'], ['DL', 'UL'],
592                                              ['high', 'low'], 'orientation',
593                                              list(range(0, 360, 10)))
594
595
596class WifiOtaThroughputStability_45Degree_Test(WifiOtaThroughputStabilityTest):
597    def __init__(self, controllers):
598        WifiOtaThroughputStabilityTest.__init__(self, controllers)
599        self.tests = self.generate_test_cases([6, 36, 149], ['VHT20', 'VHT80'],
600                                              ['TCP'], ['DL', 'UL'],
601                                              ['high', 'low'], 'orientation',
602                                              list(range(0, 360, 45)))
603
604
605class WifiOtaThroughputStability_SteppedStirrers_Test(
606        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'],
612                                              'stepped stirrers',
613                                              list(range(100)))
614