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