• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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 logging
18import queue
19import threading
20import time
21
22import acts.base_test
23import acts_contrib.test_utils.wifi.wifi_test_utils as wutils
24import acts_contrib.test_utils.tel.tel_test_utils as tutils
25
26from acts import asserts
27from acts import signals
28from acts import utils
29from acts.controllers.iperf_server import IPerfServer
30from acts.test_decorators import test_tracker_info
31from acts_contrib.test_utils.bt.bt_test_utils import enable_bluetooth
32from acts_contrib.test_utils.bt.bt_test_utils import disable_bluetooth
33from acts_contrib.test_utils.wifi.WifiBaseTest import WifiBaseTest
34
35WifiEnums = wutils.WifiEnums
36
37WAIT_FOR_AUTO_CONNECT = 40
38WAIT_WIFI_DISCONNECT_SEC = 60
39WAIT_BEFORE_CONNECTION = 30
40DEFAULT_TIMEOUT = 10
41PING_ADDR = 'www.google.com'
42BAND_2GHZ = 0
43BAND_5GHZ = 1
44MIN_ATTN = 0
45MAX_ATTN = 95
46
47
48class WifiStressTest(WifiBaseTest):
49    """WiFi Stress test class.
50
51    Test Bed Requirement:
52    * Two Android device
53    * Several Wi-Fi networks visible to the device, including an open Wi-Fi
54      network.
55    """
56
57    def __init__(self, configs):
58        super().__init__(configs)
59        self.enable_packet_log = True
60
61    def setup_class(self):
62        super().setup_class()
63
64        self.dut = self.android_devices[0]
65        # Note that test_stress_softAP_startup_and_stop_5g will always fail
66        # when testing with a single device.
67        if len(self.android_devices) > 1:
68            self.dut_client = self.android_devices[1]
69        else:
70            self.dut_client = None
71        wutils.wifi_test_device_init(self.dut)
72        req_params = []
73        opt_param = [
74            "open_network", "reference_networks", "iperf_server_address",
75            "stress_count", "stress_hours", "attn_vals", "pno_interval",
76        ]
77        self.unpack_userparams(req_param_names=req_params,
78                               opt_param_names=opt_param)
79
80        if "AccessPoint" in self.user_params:
81            self.legacy_configure_ap_and_start(ap_count=2)
82        elif "OpenWrtAP" in self.user_params:
83            self.configure_openwrt_ap_and_start(open_network=True,
84                                                wpa_network=True,
85                                                ap_count=2)
86        asserts.assert_true(
87            len(self.reference_networks) > 0,
88            "Need at least one reference network with psk.")
89        self.wpa_2g = self.reference_networks[0]["2g"]
90        self.wpa_5g = self.reference_networks[0]["5g"]
91        self.open_2g = self.open_network[0]["2g"]
92        self.open_5g = self.open_network[0]["5g"]
93        self.networks = [self.wpa_2g, self.wpa_5g, self.open_2g, self.open_5g]
94
95        # Use local host as iperf server.
96        asserts.assert_true(
97          wutils.get_host_public_ipv4_address(),
98          "The host has no public ip address")
99        self.iperf_server_address = wutils.get_host_public_ipv4_address()
100        self.iperf_server_port = wutils.get_iperf_server_port()
101        try:
102          self.iperf_server = IPerfServer(self.iperf_server_port)
103          self.iperf_server.start()
104          logging.info(f"IPerf server started on {self.iperf_server_port}")
105        except Exception as e:
106          raise signals.TestFailure("Failed to start iperf3 server: %s" % e)
107
108    def setup_test(self):
109        super().setup_test()
110        self.dut.droid.wakeLockAcquireBright()
111        self.dut.droid.wakeUpNow()
112
113    def teardown_test(self):
114        super().teardown_test()
115        if self.dut.droid.wifiIsApEnabled():
116            wutils.stop_wifi_tethering(self.dut)
117        self.dut.droid.wakeLockRelease()
118        self.dut.droid.goToSleepNow()
119        wutils.reset_wifi(self.dut)
120        self.log.debug("Toggling Airplane mode OFF")
121        asserts.assert_true(
122            acts.utils.force_airplane_mode(self.dut, False),
123            "Can not turn airplane mode off: %s" % self.dut.serial)
124
125    def teardown_class(self):
126        wutils.reset_wifi(self.dut)
127        self.iperf_server.stop()
128        if "AccessPoint" in self.user_params:
129            del self.user_params["reference_networks"]
130            del self.user_params["open_network"]
131
132    """Helper Functions"""
133
134    def scan_and_connect_by_ssid(self, ad, network):
135        """Scan for network and connect using network information.
136
137        Args:
138            network: A dictionary representing the network to connect to.
139
140        """
141        ssid = network[WifiEnums.SSID_KEY]
142        wutils.start_wifi_connection_scan_and_ensure_network_found(ad, ssid)
143        wutils.wifi_connect(ad, network, num_of_tries=3)
144
145    def scan_and_connect_by_id(self, network, net_id):
146        """Scan for network and connect using network id.
147
148        Args:
149            net_id: Integer specifying the network id of the network.
150
151        """
152        ssid = network[WifiEnums.SSID_KEY]
153        wutils.start_wifi_connection_scan_and_ensure_network_found(
154            self.dut, ssid)
155        wutils.wifi_connect_by_id(self.dut, net_id)
156
157    def run_ping(self, sec):
158        """Run ping for given number of seconds.
159
160        Args:
161            sec: Time in seconds to run teh ping traffic.
162
163        """
164        self.log.info("Running ping for %d seconds" % sec)
165        result = self.dut.adb.shell("ping -w %d %s" % (sec, PING_ADDR),
166                                    timeout=sec + 1)
167        self.log.debug("Ping Result = %s" % result)
168        if "100% packet loss" in result:
169            raise signals.TestFailure("100% packet loss during ping")
170
171    def start_youtube_video(self, url=None, secs=60):
172        """Start a youtube video and check if it's playing.
173
174        Args:
175            url: The URL of the youtube video to play.
176            secs: Time to play video in seconds.
177
178        """
179        ad = self.dut
180        ad.log.info("Start a youtube video")
181        ad.ensure_screen_on()
182        video_played = False
183        for count in range(2):
184            ad.unlock_screen()
185            ad.adb.shell('am start -a android.intent.action.VIEW -d "%s"' %
186                         url)
187            if tutils.wait_for_state(ad.droid.audioIsMusicActive, True, 15, 1):
188                ad.log.info("Started a video in youtube.")
189                # Play video for given seconds.
190                time.sleep(secs)
191                video_played = True
192                break
193        if not video_played:
194            raise signals.TestFailure(
195                "Youtube video did not start. Current WiFi "
196                "state is %d" % self.dut.droid.wifiCheckState())
197
198    def add_networks(self, ad, networks):
199        """Add Wi-Fi networks to an Android device and verify the networks were
200        added correctly.
201
202        Args:
203            ad: the AndroidDevice object to add networks to.
204            networks: a list of dicts, each dict represents a Wi-Fi network.
205        """
206        for network in networks:
207            ret = ad.droid.wifiAddNetwork(network)
208            asserts.assert_true(ret != -1,
209                                "Failed to add network %s" % network)
210            ad.droid.wifiEnableNetwork(ret, 0)
211        configured_networks = ad.droid.wifiGetConfiguredNetworks()
212        self.log.debug("Configured networks: %s", configured_networks)
213
214    def connect_and_verify_connected_ssid(self, expected_con, is_pno=False):
215        """Start a scan to get the DUT connected to an AP and verify the DUT
216        is connected to the correct SSID.
217
218        Args:
219            expected_con: The expected info of the network to we expect the DUT
220                to roam to.
221        """
222        if is_pno:
223            self.dut.log.info("Wait %ss for PNO to trigger.", self.pno_interval)
224            self.dut.log.info("Move the DUT in range of %s", self.wpa_2g[WifiEnums.SSID_KEY])
225            self.attenuators[0].set_atten(MIN_ATTN)
226            time.sleep(self.pno_interval)
227            wutils.wait_for_connect(self.dut, self.wpa_2g[WifiEnums.SSID_KEY])
228            connection_info = self.dut.droid.wifiGetConnectionInfo()
229            self.log.info("Triggering network selection from %s to %s",
230                      connection_info[WifiEnums.SSID_KEY],
231                      expected_con[WifiEnums.SSID_KEY])
232        else:
233            logging.info("Move the DUT in WiFi range.")
234            self.attenuators[0].set_atten(MIN_ATTN)
235            # force start a single scan so we don't have to wait for the scheduled scan.
236            wutils.start_wifi_connection_scan_and_return_status(self.dut)
237            self.log.info("Wait 60s for network selection.")
238            time.sleep(60)
239        try:
240            self.log.info("Connecting to %s network after network selection" %
241                          self.dut.droid.wifiGetConnectionInfo())
242            expected_ssid = expected_con[WifiEnums.SSID_KEY]
243            verify_con = {WifiEnums.SSID_KEY: expected_ssid}
244            wutils.verify_wifi_connection_info(self.dut, verify_con)
245            self.log.info(
246                "Connected to %s successfully after network selection",
247                expected_ssid)
248        finally:
249            pass
250
251    def run_long_traffic(self, sec, args, q):
252        try:
253            # Start IPerf traffic
254            self.log.info("Running iperf client {}".format(args))
255            result, data = self.dut.run_iperf_client(self.iperf_server_address,
256                                                     args,
257                                                     timeout=sec + 1)
258            if not result:
259                self.log.debug("Error occurred in iPerf traffic.")
260                self.run_ping(sec)
261            q.put(True)
262        except:
263            q.put(False)
264
265    """Tests"""
266
267    @test_tracker_info(uuid="cd0016c6-58cf-4361-b551-821c0b8d2554")
268    def test_stress_toggle_wifi_state(self):
269        """Toggle WiFi state ON and OFF for N times."""
270        for count in range(self.stress_count):
271            """Test toggling wifi"""
272            try:
273                self.log.debug("Going from on to off.")
274                wutils.wifi_toggle_state(self.dut, False)
275                self.log.debug("Going from off to on.")
276                startTime = time.time()
277                wutils.wifi_toggle_state(self.dut, True)
278                startup_time = time.time() - startTime
279                self.log.debug("WiFi was enabled on the device in %s s." %
280                               startup_time)
281            except:
282                signals.TestFailure(details="",
283                                    extras={
284                                        "Iterations": "%d" % self.stress_count,
285                                        "Pass": "%d" % count
286                                    })
287        raise signals.TestPass(details="",
288                               extras={
289                                   "Iterations": "%d" % self.stress_count,
290                                   "Pass": "%d" % (count + 1)
291                               })
292
293    @test_tracker_info(uuid="4e591cec-9251-4d52-bc6e-6621507524dc")
294    def test_stress_toggle_wifi_state_bluetooth_on(self):
295        """Toggle WiFi state ON and OFF for N times when bluetooth ON."""
296        enable_bluetooth(self.dut.droid, self.dut.ed)
297        for count in range(self.stress_count):
298            """Test toggling wifi"""
299            try:
300                self.log.debug("Going from on to off.")
301                wutils.wifi_toggle_state(self.dut, False)
302                self.log.debug("Going from off to on.")
303                startTime = time.time()
304                wutils.wifi_toggle_state(self.dut, True)
305                startup_time = time.time() - startTime
306                self.log.debug("WiFi was enabled on the device in %s s." %
307                               startup_time)
308            except:
309                signals.TestFailure(details="",
310                                    extras={
311                                        "Iterations": "%d" % self.stress_count,
312                                        "Pass": "%d" % count
313                                    })
314        disable_bluetooth(self.dut.droid)
315        raise signals.TestPass(details="",
316                               extras={
317                                   "Iterations": "%d" % self.stress_count,
318                                   "Pass": "%d" % (count + 1)
319                               })
320
321    @test_tracker_info(uuid="49e3916a-9580-4bf7-a60d-a0f2545dcdde")
322    def test_stress_connect_traffic_disconnect_5g(self):
323        """Test to connect and disconnect from a network for N times.
324
325           Steps:
326               1. Scan and connect to a network.
327               2. Run IPerf to upload data for few seconds.
328               3. Disconnect.
329               4. Repeat 1-3.
330
331        """
332        for count in range(self.stress_count):
333            try:
334                net_id = self.dut.droid.wifiAddNetwork(self.wpa_5g)
335                asserts.assert_true(net_id != -1,
336                                    "Add network %r failed" % self.wpa_5g)
337                self.scan_and_connect_by_id(self.wpa_5g, net_id)
338                # Start IPerf traffic from phone to server.
339                # Upload data for 10s.
340                args = "-p {} -t {}".format(self.iperf_server_port, 10)
341                self.log.info("Running iperf client {}".format(args))
342                result, data = self.dut.run_iperf_client(
343                    self.iperf_server_address, args)
344                if not result:
345                    self.log.debug("Error occurred in iPerf traffic.")
346                    self.run_ping(10)
347                wutils.wifi_forget_network(self.dut,
348                                           self.wpa_5g[WifiEnums.SSID_KEY])
349                time.sleep(WAIT_BEFORE_CONNECTION)
350            except:
351                raise signals.TestFailure(
352                    "Network connect-disconnect failed."
353                    "Look at logs",
354                    extras={
355                        "Iterations": "%d" % self.stress_count,
356                        "Pass": "%d" % count
357                    })
358        raise signals.TestPass(details="",
359                               extras={
360                                   "Iterations": "%d" % self.stress_count,
361                                   "Pass": "%d" % (count + 1)
362                               })
363
364    @test_tracker_info(uuid="e9827dff-0755-43ec-8b50-1f9756958460")
365    def test_stress_connect_long_traffic_5g(self):
366        """Test to connect to network and hold connection for few hours.
367
368           Steps:
369               1. Scan and connect to a network.
370               2. Run IPerf to download data for few hours.
371               3. Run IPerf to upload data for few hours.
372               4. Verify no WiFi disconnects/data interruption.
373
374        """
375        self.scan_and_connect_by_ssid(self.dut, self.wpa_5g)
376        self.scan_and_connect_by_ssid(self.dut_client, self.wpa_5g)
377
378        q = queue.Queue()
379        sec = self.stress_hours * 60 * 60
380        start_time = time.time()
381
382        dl_args = "-p {} -t {} -b1M -R".format(self.iperf_server_port, sec)
383        dl = threading.Thread(target=self.run_long_traffic,
384                              args=(sec, dl_args, q))
385        dl.start()
386        dl.join()
387
388        total_time = time.time() - start_time
389        self.log.debug("WiFi state = %d" % self.dut.droid.wifiCheckState())
390        while (q.qsize() > 0):
391            if not q.get():
392                raise signals.TestFailure("Network long-connect failed.",
393                                          extras={
394                                              "Total Hours":
395                                              "%d" % self.stress_hours,
396                                              "Seconds Run": "%d" % total_time
397                                          })
398        raise signals.TestPass(details="",
399                               extras={
400                                   "Total Hours": "%d" % self.stress_hours,
401                                   "Seconds Run": "%d" % total_time
402                               })
403
404    @test_tracker_info(uuid="591d257d-9477-4a89-a220-5715c93a76a7")
405    def test_stress_youtube_5g(self):
406        """Test to connect to network and play various youtube videos.
407
408           Steps:
409               1. Scan and connect to a network.
410               2. Loop through and play a list of youtube videos.
411               3. Verify no WiFi disconnects/data interruption.
412
413        """
414        # List of Youtube 4K videos.
415        videos = [
416            "https://www.youtube.com/watch?v=TKmGU77INaM",
417            "https://www.youtube.com/watch?v=WNCl-69POro",
418            "https://www.youtube.com/watch?v=dVkK36KOcqs",
419            "https://www.youtube.com/watch?v=0wCC3aLXdOw",
420            "https://www.youtube.com/watch?v=QpyGNwnEmKo",
421            "https://www.youtube.com/watch?v=RK1K2bCg4J8"
422        ]
423        try:
424            self.scan_and_connect_by_ssid(self.dut, self.wpa_5g)
425            start_time = time.time()
426            for video in videos:
427                self.start_youtube_video(url=video, secs=10 * 60)
428        except:
429            total_time = time.time() - start_time
430            raise signals.TestFailure("The youtube stress test has failed."
431                                      "WiFi State = %d" %
432                                      self.dut.droid.wifiCheckState(),
433                                      extras={
434                                          "Total Hours": "1",
435                                          "Seconds Run": "%d" % total_time
436                                      })
437        total_time = time.time() - start_time
438        self.log.debug("WiFi state = %d" % self.dut.droid.wifiCheckState())
439        raise signals.TestPass(details="",
440                               extras={
441                                   "Total Hours": "1",
442                                   "Seconds Run": "%d" % total_time
443                               })
444
445    @test_tracker_info(uuid="d367c83e-5b00-4028-9ed8-f7b875997d13")
446    def test_stress_wifi_failover(self):
447        """This test does aggressive failover to several networks in list.
448
449           Steps:
450               1. Add and enable few networks.
451               2. Let device auto-connect.
452               3. Remove the connected network.
453               4. Repeat 2-3.
454               5. Device should connect to a network until all networks are
455                  exhausted.
456
457        """
458        for count in range(int(self.stress_count / 4)):
459            wutils.reset_wifi(self.dut)
460            ssids = list()
461            for network in self.networks:
462                ssids.append(network[WifiEnums.SSID_KEY])
463                ret = self.dut.droid.wifiAddNetwork(network)
464                asserts.assert_true(ret != -1,
465                                    "Add network %r failed" % network)
466                self.dut.droid.wifiEnableNetwork(ret, 0)
467            self.dut.droid.wifiStartScan()
468            time.sleep(WAIT_FOR_AUTO_CONNECT)
469            cur_network = self.dut.droid.wifiGetConnectionInfo()
470            cur_ssid = cur_network[WifiEnums.SSID_KEY]
471            self.log.info("Cur_ssid = %s" % cur_ssid)
472            for i in range(0, len(self.networks)):
473                self.log.debug("Forget network %s" % cur_ssid)
474                wutils.wifi_forget_network(self.dut, cur_ssid)
475                time.sleep(WAIT_FOR_AUTO_CONNECT)
476                cur_network = self.dut.droid.wifiGetConnectionInfo()
477                cur_ssid = cur_network[WifiEnums.SSID_KEY]
478                self.log.info("Cur_ssid = %s" % cur_ssid)
479                if i == len(self.networks) - 1:
480                    break
481                if cur_ssid not in ssids:
482                    raise signals.TestFailure("Device did not failover to the "
483                                              "expected network. SSID = %s" %
484                                              cur_ssid)
485            network_config = self.dut.droid.wifiGetConfiguredNetworks()
486            self.log.info("Network Config = %s" % network_config)
487            if len(network_config):
488                raise signals.TestFailure(
489                    "All the network configurations were not "
490                    "removed. Configured networks = %s" % network_config,
491                    extras={
492                        "Iterations": "%d" % self.stress_count,
493                        "Pass": "%d" % (count * 4)
494                    })
495        raise signals.TestPass(details="",
496                               extras={
497                                   "Iterations": "%d" % self.stress_count,
498                                   "Pass": "%d" % ((count + 1) * 4)
499                               })
500
501    @test_tracker_info(uuid="2c19e8d1-ac16-4d7e-b309-795144e6b956")
502    def test_stress_softAP_startup_and_stop_5g(self):
503        """Test to bring up softAP and down for N times.
504
505        Steps:
506            1. Bring up softAP on 5G.
507            2. Check for softAP on teh client device.
508            3. Turn ON WiFi.
509            4. Verify softAP is turned down and WiFi is up.
510
511        """
512        ap_ssid = "softap_" + utils.rand_ascii_str(8)
513        ap_password = utils.rand_ascii_str(8)
514        self.dut.log.info("softap setup: %s %s", ap_ssid, ap_password)
515        config = {wutils.WifiEnums.SSID_KEY: ap_ssid}
516        config[wutils.WifiEnums.PWD_KEY] = ap_password
517        # Set country code explicitly to "US".
518        wutils.set_wifi_country_code(self.dut, wutils.WifiEnums.CountryCode.US)
519        wutils.set_wifi_country_code(self.dut_client,
520                                     wutils.WifiEnums.CountryCode.US)
521        for count in range(self.stress_count):
522            initial_wifi_state = self.dut.droid.wifiCheckState()
523            wutils.start_wifi_tethering(self.dut, ap_ssid, ap_password,
524                                        WifiEnums.WIFI_CONFIG_APBAND_5G)
525            wutils.start_wifi_connection_scan_and_ensure_network_found(
526                self.dut_client, ap_ssid)
527            wutils.stop_wifi_tethering(self.dut)
528            asserts.assert_false(self.dut.droid.wifiIsApEnabled(),
529                                 "SoftAp failed to shutdown!")
530            # Give some time for WiFi to come back to previous state.
531            time.sleep(2)
532            cur_wifi_state = self.dut.droid.wifiCheckState()
533            if initial_wifi_state != cur_wifi_state:
534                raise signals.TestFailure(
535                    "Wifi state was %d before softAP and %d now!" %
536                    (initial_wifi_state, cur_wifi_state),
537                    extras={
538                        "Iterations": "%d" % self.stress_count,
539                        "Pass": "%d" % count
540                    })
541        raise signals.TestPass(details="",
542                               extras={
543                                   "Iterations": "%d" % self.stress_count,
544                                   "Pass": "%d" % (count + 1)
545                               })
546
547    @test_tracker_info(uuid="eb22e26b-95d1-4580-8c76-85dfe6a42a0f")
548    def test_stress_wifi_roaming(self):
549        AP1_network = self.reference_networks[0]["5g"]
550        AP2_network = self.reference_networks[1]["5g"]
551        wutils.set_attns(self.attenuators, "AP1_on_AP2_off")
552        self.scan_and_connect_by_ssid(self.dut, AP1_network)
553        # Reduce iteration to half because each iteration does two roams.
554        for count in range(int(self.stress_count / 2)):
555            self.log.info("Roaming iteration %d, from %s to %s", count,
556                          AP1_network, AP2_network)
557            try:
558                wutils.trigger_roaming_and_validate(self.dut, self.attenuators,
559                                                    "AP1_off_AP2_on",
560                                                    AP2_network)
561                self.log.info("Roaming iteration %d, from %s to %s", count,
562                              AP2_network, AP1_network)
563                wutils.trigger_roaming_and_validate(self.dut, self.attenuators,
564                                                    "AP1_on_AP2_off",
565                                                    AP1_network)
566            except:
567                raise signals.TestFailure("Roaming failed. Look at logs",
568                                          extras={
569                                              "Iterations":
570                                              "%d" % self.stress_count,
571                                              "Pass": "%d" % (count * 2)
572                                          })
573        raise signals.TestPass(details="",
574                               extras={
575                                   "Iterations": "%d" % self.stress_count,
576                                   "Pass": "%d" % ((count + 1) * 2)
577                               })
578
579    @test_tracker_info(uuid="e8ae8cd2-c315-4c08-9eb3-83db65b78a58")
580    def test_stress_network_selector_2G_connection(self):
581        """
582            1. Add one saved 2G network to DUT.
583            2. Move the DUT in range.
584            3. Verify the DUT is connected to the network.
585            4. Move the DUT out of range
586            5. Repeat step 2-4
587        """
588        for attenuator in self.attenuators:
589            attenuator.set_atten(95)
590        # add a saved network to DUT
591        networks = [self.reference_networks[0]['2g']]
592        self.add_networks(self.dut, networks)
593        for count in range(self.stress_count):
594            self.connect_and_verify_connected_ssid(
595                self.reference_networks[0]['2g'])
596            logging.info("Move the DUT out of WiFi range and wait 10 seconds.")
597            self.attenuators[0].set_atten(95)
598            time.sleep(10)
599        wutils.set_attns(self.attenuators, "default")
600        raise signals.TestPass(details="",
601                               extras={
602                                   "Iterations": "%d" % self.stress_count,
603                                   "Pass": "%d" % (count + 1)
604                               })
605
606    @test_tracker_info(uuid="5d5d14cb-3cd1-4b3d-8c04-0d6f4b764b6b")
607    def test_stress_pno_connection_to_2g(self):
608        """Test PNO triggered autoconnect to a network for N times
609
610        Steps:
611        1. DUT Connects to a 2GHz network.
612        2. Screen off DUT.
613        3. Attenuate 2GHz network and wait for triggering PNO.
614        4. Move the DUT in range of 2GHz network.
615        5. Check the device connected to 2GHz network automatically.
616        6. Repeat step 3-5.
617        """
618        self.scan_and_connect_by_ssid(self.dut, self.wpa_2g)
619        for attenuator in self.attenuators:
620            attenuator.set_atten(MAX_ATTN)
621        self.dut.droid.wakeLockRelease()
622        self.dut.droid.goToSleepNow()
623        self.dut.log.info("DUT screen off")
624        for count in range(self.stress_count):
625            self.connect_and_verify_connected_ssid(
626                self.reference_networks[0]['2g'], is_pno=True)
627            logging.info("Move the DUT out of range by fully attenuate %s",
628                         self.wpa_2g[WifiEnums.SSID_KEY])
629            self.attenuators[0].set_atten(MAX_ATTN)
630            logging.info("Wait for DUT to disconnect from %s",
631                         self.wpa_2g[WifiEnums.SSID_KEY])
632            wutils.wait_for_disconnect(self.dut,
633                                       timeout=WAIT_WIFI_DISCONNECT_SEC)
634            self.dut.log.info("DUT disconnected from from %s",
635                              self.wpa_2g[WifiEnums.SSID_KEY])
636        wutils.set_attns(self.attenuators, "default")
637        raise signals.TestPass(details="",
638                               extras={
639                                   "Iterations": "%d" % self.stress_count,
640                                   "Pass": "%d" % (count + 1)
641                               })
642
643    @test_tracker_info(uuid="c880e742-8d20-4134-b717-5b6d45f6c337")
644    def test_2g_sta_wifi_on_off_under_airplane_mode(self):
645        """Toggle WiFi state ON and OFF for N times when airplane mode ON."""
646        self.scan_and_connect_by_ssid(self.dut, self.wpa_2g)
647        self.log.debug("Toggling Airplane mode ON")
648        asserts.assert_true(
649            acts.utils.force_airplane_mode(self.dut, True),
650            "Can not turn on airplane mode on: %s" % self.dut.serial)
651        time.sleep(DEFAULT_TIMEOUT)
652        for count in range(self.stress_count):
653            """Test toggling wifi"""
654            try:
655                self.log.debug("Going from on to off.")
656                wutils.wifi_toggle_state(self.dut, False)
657                self.log.debug("Going from off to on.")
658                startTime = time.time()
659                wutils.wifi_toggle_state(self.dut, True)
660                startup_time = time.time() - startTime
661                self.log.debug("WiFi was enabled on the device in %s s." %
662                               startup_time)
663                time.sleep(DEFAULT_TIMEOUT)
664                # Start IPerf traffic from phone to server.
665                # Upload data for 10s.
666                args = "-p {} -t {}".format(self.iperf_server_port, 10)
667                self.log.info("Running iperf client {}".format(args))
668                result, data = self.dut.run_iperf_client(
669                    self.iperf_server_address, args)
670                if not result:
671                    self.log.debug("Error occurred in iPerf traffic.")
672                    self.run_ping(10)
673            except:
674                signals.TestFailure(details="",
675                                    extras={
676                                        "Iterations": "%d" % self.stress_count,
677                                        "Pass": "%d" % count
678                                    })
679        raise signals.TestPass(details="",
680                               extras={
681                                   "Iterations": "%d" % self.stress_count,
682                                   "Pass": "%d" % (count + 1)
683                               })
684