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 pprint
18import time
19
20import acts.base_test
21import acts.test_utils.wifi.wifi_test_utils as wutils
22import acts.utils
23
24from acts import asserts
25from acts import signals
26from acts import utils
27from acts.test_decorators import test_tracker_info
28from acts.test_utils.wifi.WifiBaseTest import WifiBaseTest
29WifiEnums = wutils.WifiEnums
30
31WAIT_FOR_AUTO_CONNECT = 40
32WAIT_BEFORE_CONNECTION = 30
33
34TIMEOUT = 1
35PING_ADDR = 'www.google.com'
36
37class WifiStressTest(WifiBaseTest):
38    """WiFi Stress test class.
39
40    Test Bed Requirement:
41    * Two Android device
42    * Several Wi-Fi networks visible to the device, including an open Wi-Fi
43      network.
44    """
45
46    def __init__(self, controllers):
47        WifiBaseTest.__init__(self, controllers)
48
49    def setup_class(self):
50        self.dut = self.android_devices[0]
51        self.dut_client = self.android_devices[1]
52        wutils.wifi_test_device_init(self.dut)
53        req_params = []
54        opt_param = [
55            "open_network", "reference_networks", "iperf_server_address",
56            "stress_count", "stress_hours"]
57        self.unpack_userparams(
58            req_param_names=req_params, opt_param_names=opt_param)
59
60        if "AccessPoint" in self.user_params:
61            self.legacy_configure_ap_and_start(ap_count=2)
62
63        asserts.assert_true(
64            len(self.reference_networks) > 0,
65            "Need at least one reference network with psk.")
66        self.wpa_2g = self.reference_networks[0]["2g"]
67        self.wpa_5g = self.reference_networks[0]["5g"]
68        self.open_2g = self.open_network[0]["2g"]
69        self.open_5g = self.open_network[0]["5g"]
70        self.networks = [self.wpa_2g, self.wpa_5g, self.open_2g, self.open_5g]
71        if "iperf_server_address" in self.user_params:
72            self.iperf_server = self.iperf_servers[0]
73        if hasattr(self, 'iperf_server'):
74            self.iperf_server.start()
75
76    def setup_test(self):
77        self.dut.droid.wakeLockAcquireBright()
78        self.dut.droid.wakeUpNow()
79
80    def teardown_test(self):
81        self.dut.droid.wakeLockRelease()
82        self.dut.droid.goToSleepNow()
83        wutils.reset_wifi(self.dut)
84
85    def on_fail(self, test_name, begin_time):
86        self.dut.take_bug_report(test_name, begin_time)
87        self.dut.cat_adb_log(test_name, begin_time)
88
89    def teardown_class(self):
90        wutils.reset_wifi(self.dut)
91        if hasattr(self, 'iperf_server'):
92            self.iperf_server.stop()
93        if "AccessPoint" in self.user_params:
94            del self.user_params["reference_networks"]
95            del self.user_params["open_network"]
96
97    """Helper Functions"""
98
99    def scan_and_connect_by_ssid(self, network):
100        """Scan for network and connect using network information.
101
102        Args:
103            network: A dictionary representing the network to connect to.
104
105        """
106        ssid = network[WifiEnums.SSID_KEY]
107        wutils.start_wifi_connection_scan_and_ensure_network_found(self.dut,
108            ssid)
109        wutils.wifi_connect(self.dut, network, num_of_tries=3)
110
111    def scan_and_connect_by_id(self, network, net_id):
112        """Scan for network and connect using network id.
113
114        Args:
115            net_id: Integer specifying the network id of the network.
116
117        """
118        ssid = network[WifiEnums.SSID_KEY]
119        wutils.start_wifi_connection_scan_and_ensure_network_found(self.dut,
120            ssid)
121        wutils.wifi_connect_by_id(self.dut, net_id)
122
123    def run_ping(self, sec):
124        """Run ping for given number of seconds.
125
126        Args:
127            sec: Time in seconds to run teh ping traffic.
128
129        """
130        self.log.info("Running ping for %d seconds" % sec)
131        result = self.dut.adb.shell("ping -w %d %s" %(sec, PING_ADDR),
132            timeout=sec+1)
133        self.log.debug("Ping Result = %s" % result)
134        if "100% packet loss" in result:
135            raise signals.TestFailure("100% packet loss during ping")
136
137    """Tests"""
138
139    @test_tracker_info(uuid="cd0016c6-58cf-4361-b551-821c0b8d2554")
140    def test_stress_toggle_wifi_state(self):
141        """Toggle WiFi state ON and OFF for N times."""
142        for count in range(self.stress_count):
143            """Test toggling wifi"""
144            try:
145                self.log.debug("Going from on to off.")
146                wutils.wifi_toggle_state(self.dut, False)
147                self.log.debug("Going from off to on.")
148                startTime = time.time()
149                wutils.wifi_toggle_state(self.dut, True)
150                startup_time = time.time() - startTime
151                self.log.debug("WiFi was enabled on the device in %s s." %
152                    startup_time)
153            except:
154                signals.TestFailure(details="", extras={"Iterations":"%d" %
155                    self.stress_count, "Pass":"%d" %count})
156        raise signals.TestPass(details="", extras={"Iterations":"%d" %
157            self.stress_count, "Pass":"%d" %(count+1)})
158
159    @test_tracker_info(uuid="49e3916a-9580-4bf7-a60d-a0f2545dcdde")
160    def test_stress_connect_traffic_disconnect_5g(self):
161        """Test to connect and disconnect from a network for N times.
162
163           Steps:
164               1. Scan and connect to a network.
165               2. Run IPerf to upload data for few seconds.
166               3. Disconnect.
167               4. Repeat 1-3.
168
169        """
170        for count in range(self.stress_count):
171            try:
172                net_id = self.dut.droid.wifiAddNetwork(self.wpa_5g)
173                asserts.assert_true(net_id != -1, "Add network %r failed" % self.wpa_5g)
174                self.scan_and_connect_by_id(self.wpa_5g, net_id)
175                # Start IPerf traffic from phone to server.
176                # Upload data for 10s.
177                args = "-p {} -t {}".format(self.iperf_server.port, 10)
178                self.log.info("Running iperf client {}".format(args))
179                result, data = self.dut.run_iperf_client(self.iperf_server_address, args)
180                if not result:
181                    self.log.debug("Error occurred in iPerf traffic.")
182                    self.run_ping(10)
183                wutils.wifi_forget_network(self.dut,self.wpa_5g[WifiEnums.SSID_KEY])
184                time.sleep(WAIT_BEFORE_CONNECTION)
185            except:
186                raise signals.TestFailure("Network connect-disconnect failed."
187                    "Look at logs", extras={"Iterations":"%d" %
188                        self.stress_count, "Pass":"%d" %count})
189        raise signals.TestPass(details="", extras={"Iterations":"%d" %
190            self.stress_count, "Pass":"%d" %(count+1)})
191
192    @test_tracker_info(uuid="e9827dff-0755-43ec-8b50-1f9756958460")
193    def test_stress_connect_long_traffic_5g(self):
194        """Test to connect to network and hold connection for few hours.
195
196           Steps:
197               1. Scan and connect to a network.
198               2. Run IPerf to download data for few hours.
199               3. Verify no WiFi disconnects/data interruption.
200
201        """
202        try:
203            self.scan_and_connect_by_ssid(self.wpa_5g)
204            # Start IPerf traffic from server to phone.
205            # Download data for 5 hours.
206            sec = self.stress_hours * 60 * 60
207            args = "-p {} -t {} -R".format(self.iperf_server.port, sec)
208            self.log.info("Running iperf client {}".format(args))
209            result, data = self.dut.run_iperf_client(self.iperf_server_address,
210                args, timeout=sec+1)
211            if not result:
212                self.log.debug("Error occurred in iPerf traffic.")
213                self.run_ping(sec)
214        except:
215            raise signals.TestFailure("Network long-connect failed."
216                "Look at logs", extras={"Total Hours":"%d" %self.stress_hours,
217                    "Seconds Run":"UNKNOWN"})
218        raise signals.TestPass(details="", extras={"Total Hours":"%d" %
219            self.stress_hours, "Seconds":"%d" %sec})
220
221    @test_tracker_info(uuid="d367c83e-5b00-4028-9ed8-f7b875997d13")
222    def test_stress_wifi_failover(self):
223        """This test does aggressive failover to several networks in list.
224
225           Steps:
226               1. Add and enable few networks.
227               2. Let device auto-connect.
228               3. Remove the connected network.
229               4. Repeat 2-3.
230               5. Device should connect to a network until all networks are
231                  exhausted.
232
233        """
234        for count in range(int(self.stress_count/4)):
235            ssids = list()
236            for network in self.networks:
237                ssids.append(network[WifiEnums.SSID_KEY])
238                ret = self.dut.droid.wifiAddNetwork(network)
239                asserts.assert_true(ret != -1, "Add network %r failed" % network)
240                self.dut.droid.wifiEnableNetwork(ret, 0)
241            time.sleep(WAIT_FOR_AUTO_CONNECT)
242            cur_network = self.dut.droid.wifiGetConnectionInfo()
243            cur_ssid = cur_network[WifiEnums.SSID_KEY]
244            self.log.info("Cur_ssid = %s" % cur_ssid)
245            for i in range(0,len(self.networks)):
246                self.log.debug("Forget network %s" % cur_ssid)
247                wutils.wifi_forget_network(self.dut, cur_ssid)
248                time.sleep(WAIT_FOR_AUTO_CONNECT)
249                cur_network = self.dut.droid.wifiGetConnectionInfo()
250                cur_ssid = cur_network[WifiEnums.SSID_KEY]
251                self.log.info("Cur_ssid = %s" % cur_ssid)
252                if i == len(self.networks) - 1:
253                    break
254                if cur_ssid not in ssids:
255                    raise signals.TestFailure("Device did not failover to the "
256                        "expected network. SSID = %s" % cur_ssid)
257            network_config = self.dut.droid.wifiGetConfiguredNetworks()
258            self.log.info("Network Config = %s" % network_config)
259            if len(network_config):
260                raise signals.TestFailure("All the network configurations were not "
261                    "removed. Configured networks = %s" % network_config,
262                        extras={"Iterations":"%d" % self.stress_count,
263                            "Pass":"%d" %(count*4)})
264        raise signals.TestPass(details="", extras={"Iterations":"%d" %
265            self.stress_count, "Pass":"%d" %((count+1)*4)})
266
267    @test_tracker_info(uuid="2c19e8d1-ac16-4d7e-b309-795144e6b956")
268    def test_stress_softAP_startup_and_stop_5g(self):
269        """Test to bring up softAP and down for N times.
270
271        Steps:
272            1. Bring up softAP on 5G.
273            2. Check for softAP on teh client device.
274            3. Turn ON WiFi.
275            4. Verify softAP is turned down and WiFi is up.
276
277        """
278        # Set country code explicitly to "US".
279        self.dut.droid.wifiSetCountryCode(wutils.WifiEnums.CountryCode.US)
280        self.dut_client.droid.wifiSetCountryCode(wutils.WifiEnums.CountryCode.US)
281        ap_ssid = "softap_" + utils.rand_ascii_str(8)
282        ap_password = utils.rand_ascii_str(8)
283        self.dut.log.info("softap setup: %s %s", ap_ssid, ap_password)
284        config = {wutils.WifiEnums.SSID_KEY: ap_ssid}
285        config[wutils.WifiEnums.PWD_KEY] = ap_password
286        for count in range(self.stress_count):
287            initial_wifi_state = self.dut.droid.wifiCheckState()
288            wutils.start_wifi_tethering(self.dut,
289                ap_ssid,
290                ap_password,
291                WifiEnums.WIFI_CONFIG_APBAND_5G)
292            wutils.start_wifi_connection_scan_and_ensure_network_found(
293                self.dut_client, ap_ssid)
294            wutils.stop_wifi_tethering(self.dut)
295            asserts.assert_false(self.dut.droid.wifiIsApEnabled(),
296                                 "SoftAp failed to shutdown!")
297            time.sleep(TIMEOUT)
298            cur_wifi_state = self.dut.droid.wifiCheckState()
299            if initial_wifi_state != cur_wifi_state:
300               raise signals.TestFailure("Wifi state was %d before softAP and %d now!" %
301                    (initial_wifi_state, cur_wifi_state),
302                        extras={"Iterations":"%d" % self.stress_count,
303                            "Pass":"%d" %count})
304        raise signals.TestPass(details="", extras={"Iterations":"%d" %
305            self.stress_count, "Pass":"%d" %(count+1)})
306
307    @test_tracker_info(uuid="eb22e26b-95d1-4580-8c76-85dfe6a42a0f")
308    def test_stress_wifi_roaming(self):
309        AP1_network = self.reference_networks[0]["5g"]
310        AP2_network = self.reference_networks[1]["5g"]
311        wutils.set_attns(self.attenuators, "AP1_on_AP2_off")
312        self.scan_and_connect_by_ssid(AP1_network)
313        # Reduce iteration to half because each iteration does two roams.
314        for count in range(int(self.stress_count/2)):
315            self.log.info("Roaming iteration %d, from %s to %s", count,
316                           AP1_network, AP2_network)
317            try:
318                wutils.trigger_roaming_and_validate(self.dut, self.attenuators,
319                    "AP1_off_AP2_on", AP2_network)
320                self.log.info("Roaming iteration %d, from %s to %s", count,
321                               AP2_network, AP1_network)
322                wutils.trigger_roaming_and_validate(self.dut, self.attenuators,
323                    "AP1_on_AP2_off", AP1_network)
324            except:
325                raise signals.TestFailure("Roaming failed. Look at logs",
326                    extras={"Iterations":"%d" %self.stress_count, "Pass":"%d" %
327                        (count*2)})
328        raise signals.TestPass(details="", extras={"Iterations":"%d" %
329            self.stress_count, "Pass":"%d" %((count+1)*2)})
330
331