1#!/usr/bin/env python3.4
2#
3#   Copyright 2018 - 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 logging
19import os
20from acts import asserts
21from acts import base_test
22from acts.controllers import iperf_server as ipf
23from acts.controllers import iperf_client as ipc
24from acts.metrics.loggers.blackbox import BlackboxMappedMetricLogger
25from acts.test_utils.wifi import ota_sniffer
26from acts.test_utils.wifi import wifi_test_utils as wutils
27from acts.test_utils.wifi import wifi_performance_test_utils as wputils
28from WifiRvrTest import WifiRvrTest
29
30AccessPointTuple = collections.namedtuple(('AccessPointTuple'),
31                                          ['ap_settings'])
32
33
34class WifiSoftApRvrTest(WifiRvrTest):
35    def __init__(self, controllers):
36        base_test.BaseTestClass.__init__(self, controllers)
37        self.tests = ('test_rvr_TCP_DL_2GHz', 'test_rvr_TCP_UL_2GHz',
38                      'test_rvr_TCP_DL_5GHz', 'test_rvr_TCP_UL_5GHz')
39        self.testcase_metric_logger = (
40            BlackboxMappedMetricLogger.for_test_case())
41        self.testclass_metric_logger = (
42            BlackboxMappedMetricLogger.for_test_class())
43        self.publish_testcase_metrics = True
44
45    def setup_class(self):
46        """Initializes common test hardware and parameters.
47
48        This function initializes hardwares and compiles parameters that are
49        common to all tests in this class.
50        """
51        self.dut = self.android_devices[-1]
52        req_params = ['sap_test_params', 'testbed_params']
53        opt_params = ['golden_files_list', 'OTASniffer']
54        self.unpack_userparams(req_params, opt_params)
55        self.testclass_params = self.sap_test_params
56        self.num_atten = self.attenuators[0].instrument.num_atten
57        self.iperf_server = ipf.create([{
58            'AndroidDevice':
59            self.android_devices[0].serial,
60            'port':
61            '5201'
62        }])[0]
63        self.iperf_client = ipc.create([{
64            'AndroidDevice':
65            self.android_devices[1].serial,
66            'port':
67            '5201'
68        }])[0]
69        if hasattr(self,
70                   'OTASniffer') and self.testbed_params['sniffer_enable']:
71            self.sniffer = ota_sniffer.create(self.OTASniffer)[0]
72
73        self.log_path = os.path.join(logging.log_path, 'results')
74        os.makedirs(self.log_path, exist_ok=True)
75        if not hasattr(self, 'golden_files_list'):
76            if 'golden_results_path' in self.testbed_params:
77                self.golden_files_list = [
78                    os.path.join(self.testbed_params['golden_results_path'],
79                                 file) for file in
80                    os.listdir(self.testbed_params['golden_results_path'])
81                ]
82            else:
83                self.log.warning('No golden files found.')
84                self.golden_files_list = []
85        self.testclass_results = []
86
87        # Turn WiFi ON
88        for dev in self.android_devices:
89            wutils.wifi_toggle_state(dev, True)
90
91    def teardown_class(self):
92        # Turn WiFi OFF
93        wutils.stop_wifi_tethering(self.android_devices[0])
94        for dev in self.android_devices:
95            wutils.wifi_toggle_state(dev, False)
96        self.process_testclass_results()
97
98    def teardown_test(self):
99        self.iperf_server.stop()
100        wutils.stop_wifi_tethering(self.android_devices[0])
101
102    def get_sap_connection_info(self):
103        info = {}
104        info['client_ip_address'] = self.android_devices[
105            1].droid.connectivityGetIPv4Addresses('wlan0')[0]
106        info['ap_ip_address'] = self.android_devices[
107            0].droid.connectivityGetIPv4Addresses('wlan1')[0]
108        info['frequency'] = self.android_devices[1].adb.shell(
109            'wpa_cli status | grep freq').split('=')[1]
110        info['channel'] = wutils.WifiEnums.freq_to_channel[int(
111            info['frequency'])]
112        return info
113
114    def setup_sap_rvr_test(self, testcase_params):
115        """Function that gets devices ready for the test.
116
117        Args:
118            testcase_params: dict containing test-specific parameters
119        """
120        for dev in self.android_devices:
121            if not wputils.health_check(dev, 20):
122                asserts.skip('DUT health check failed. Skipping test.')
123        # Reset WiFi on all devices
124        for dev in self.android_devices:
125            self.dut.go_to_sleep()
126            wutils.reset_wifi(dev)
127            wutils.set_wifi_country_code(dev, wutils.WifiEnums.CountryCode.US)
128
129        # Setup Soft AP
130        sap_config = wutils.create_softap_config()
131        self.log.info('SoftAP Config: {}'.format(sap_config))
132        wutils.start_wifi_tethering(self.android_devices[0],
133                                    sap_config[wutils.WifiEnums.SSID_KEY],
134                                    sap_config[wutils.WifiEnums.PWD_KEY],
135                                    testcase_params['sap_band_enum'])
136        # Set attenuator to 0 dB
137        for attenuator in self.attenuators:
138            attenuator.set_atten(0, strict=False)
139        # Connect DUT to Network
140        testcase_params['test_network'] = {
141            'SSID': sap_config[wutils.WifiEnums.SSID_KEY],
142            'password': sap_config[wutils.WifiEnums.PWD_KEY]
143        }
144        wutils.wifi_connect(self.android_devices[1],
145                            testcase_params['test_network'],
146                            num_of_tries=5,
147                            check_connectivity=False)
148        # Compile meta data
149        self.access_point = AccessPointTuple(sap_config)
150        testcase_params['connection_info'] = self.get_sap_connection_info()
151        testcase_params['channel'] = testcase_params['connection_info'][
152            'channel']
153        testcase_params['test_network']['channel'] = testcase_params[
154            'connection_info']['channel']
155        if testcase_params['channel'] < 13:
156            testcase_params['mode'] = 'VHT20'
157        else:
158            testcase_params['mode'] = 'VHT80'
159        testcase_params['iperf_server_address'] = testcase_params[
160            'connection_info']['ap_ip_address']
161
162    def compile_test_params(self, testcase_params):
163        """Function that completes all test params based on the test name.
164
165        Args:
166            testcase_params: dict containing test-specific parameters
167        """
168        num_atten_steps = int((self.testclass_params['atten_stop'] -
169                               self.testclass_params['atten_start']) /
170                              self.testclass_params['atten_step'])
171        testcase_params['atten_range'] = [
172            self.testclass_params['atten_start'] +
173            x * self.testclass_params['atten_step']
174            for x in range(0, num_atten_steps)
175        ]
176
177        if testcase_params['traffic_direction'] == 'DL':
178            testcase_params['iperf_args'] = wputils.get_iperf_arg_string(
179                duration=self.testclass_params['iperf_duration'],
180                reverse_direction=1,
181                traffic_type=testcase_params['traffic_type'])
182            testcase_params['use_client_output'] = True
183        else:
184            testcase_params['iperf_args'] = wputils.get_iperf_arg_string(
185                duration=self.testclass_params['iperf_duration'],
186                reverse_direction=0,
187                traffic_type=testcase_params['traffic_type'])
188            testcase_params['use_client_output'] = False
189        return testcase_params
190
191    def _test_sap_rvr(self, testcase_params):
192        """ Function that gets called for each test case
193
194        Args:
195            testcase_params: dict containing test-specific parameters
196        """
197        # Compile test parameters from config and test name
198        testcase_params = self.compile_test_params(testcase_params)
199
200        self.setup_sap_rvr_test(testcase_params)
201        result = self.run_rvr_test(testcase_params)
202        self.testclass_results.append(result)
203        self.process_test_results(result)
204        self.pass_fail_check(result)
205
206    #Test cases
207    def test_rvr_TCP_DL_2GHz(self):
208        testcase_params = collections.OrderedDict(
209            sap_band='2GHz',
210            sap_band_enum=wutils.WifiEnums.WIFI_CONFIG_APBAND_2G,
211            traffic_type='TCP',
212            traffic_direction='DL')
213        self._test_sap_rvr(testcase_params)
214
215    def test_rvr_TCP_UL_2GHz(self):
216        testcase_params = collections.OrderedDict(
217            sap_band='2GHz',
218            sap_band_enum=wutils.WifiEnums.WIFI_CONFIG_APBAND_2G,
219            traffic_type='TCP',
220            traffic_direction='UL')
221        self._test_sap_rvr(testcase_params)
222
223    def test_rvr_TCP_DL_5GHz(self):
224        testcase_params = collections.OrderedDict(
225            sap_band='5GHz',
226            sap_band_enum=wutils.WifiEnums.WIFI_CONFIG_APBAND_5G,
227            traffic_type='TCP',
228            traffic_direction='DL')
229        self._test_sap_rvr(testcase_params)
230
231    def test_rvr_TCP_UL_5GHz(self):
232        testcase_params = collections.OrderedDict(
233            sap_band='5GHz',
234            sap_band_enum=wutils.WifiEnums.WIFI_CONFIG_APBAND_5G,
235            traffic_type='TCP',
236            traffic_direction='UL')
237        self._test_sap_rvr(testcase_params)
238