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