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