| #!/usr/bin/env python3.4 |
| # |
| # Copyright 2018 - The Android Open Source Project |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| |
| import pprint |
| import time |
| |
| import acts.base_test |
| import acts.test_utils.wifi.wifi_test_utils as wutils |
| import acts.utils |
| |
| from acts import asserts |
| from acts import signals |
| from acts import utils |
| from acts.test_decorators import test_tracker_info |
| from acts.test_utils.wifi.WifiBaseTest import WifiBaseTest |
| WifiEnums = wutils.WifiEnums |
| |
| WAIT_FOR_AUTO_CONNECT = 40 |
| WAIT_BEFORE_CONNECTION = 30 |
| |
| TIMEOUT = 1 |
| PING_ADDR = 'www.google.com' |
| |
| class WifiStressTest(WifiBaseTest): |
| """WiFi Stress test class. |
| |
| Test Bed Requirement: |
| * Two Android device |
| * Several Wi-Fi networks visible to the device, including an open Wi-Fi |
| network. |
| """ |
| |
| def __init__(self, controllers): |
| WifiBaseTest.__init__(self, controllers) |
| |
| def setup_class(self): |
| self.dut = self.android_devices[0] |
| self.dut_client = self.android_devices[1] |
| wutils.wifi_test_device_init(self.dut) |
| req_params = [] |
| opt_param = [ |
| "open_network", "reference_networks", "iperf_server_address", |
| "stress_count", "stress_hours"] |
| self.unpack_userparams( |
| req_param_names=req_params, opt_param_names=opt_param) |
| |
| if "AccessPoint" in self.user_params: |
| self.legacy_configure_ap_and_start(ap_count=2) |
| |
| asserts.assert_true( |
| len(self.reference_networks) > 0, |
| "Need at least one reference network with psk.") |
| self.wpa_2g = self.reference_networks[0]["2g"] |
| self.wpa_5g = self.reference_networks[0]["5g"] |
| self.open_2g = self.open_network[0]["2g"] |
| self.open_5g = self.open_network[0]["5g"] |
| self.networks = [self.wpa_2g, self.wpa_5g, self.open_2g, self.open_5g] |
| if "iperf_server_address" in self.user_params: |
| self.iperf_server = self.iperf_servers[0] |
| if hasattr(self, 'iperf_server'): |
| self.iperf_server.start() |
| |
| def setup_test(self): |
| self.dut.droid.wakeLockAcquireBright() |
| self.dut.droid.wakeUpNow() |
| |
| def teardown_test(self): |
| self.dut.droid.wakeLockRelease() |
| self.dut.droid.goToSleepNow() |
| wutils.reset_wifi(self.dut) |
| |
| def on_fail(self, test_name, begin_time): |
| self.dut.take_bug_report(test_name, begin_time) |
| self.dut.cat_adb_log(test_name, begin_time) |
| |
| def teardown_class(self): |
| wutils.reset_wifi(self.dut) |
| if hasattr(self, 'iperf_server'): |
| self.iperf_server.stop() |
| if "AccessPoint" in self.user_params: |
| del self.user_params["reference_networks"] |
| del self.user_params["open_network"] |
| |
| """Helper Functions""" |
| |
| def scan_and_connect_by_ssid(self, network): |
| """Scan for network and connect using network information. |
| |
| Args: |
| network: A dictionary representing the network to connect to. |
| |
| """ |
| ssid = network[WifiEnums.SSID_KEY] |
| wutils.start_wifi_connection_scan_and_ensure_network_found(self.dut, |
| ssid) |
| wutils.wifi_connect(self.dut, network, num_of_tries=3) |
| |
| def scan_and_connect_by_id(self, network, net_id): |
| """Scan for network and connect using network id. |
| |
| Args: |
| net_id: Integer specifying the network id of the network. |
| |
| """ |
| ssid = network[WifiEnums.SSID_KEY] |
| wutils.start_wifi_connection_scan_and_ensure_network_found(self.dut, |
| ssid) |
| wutils.wifi_connect_by_id(self.dut, net_id) |
| |
| def run_ping(self, sec): |
| """Run ping for given number of seconds. |
| |
| Args: |
| sec: Time in seconds to run teh ping traffic. |
| |
| """ |
| self.log.info("Running ping for %d seconds" % sec) |
| result = self.dut.adb.shell("ping -w %d %s" %(sec, PING_ADDR), |
| timeout=sec+1) |
| self.log.debug("Ping Result = %s" % result) |
| if "100% packet loss" in result: |
| raise signals.TestFailure("100% packet loss during ping") |
| |
| """Tests""" |
| |
| @test_tracker_info(uuid="cd0016c6-58cf-4361-b551-821c0b8d2554") |
| def test_stress_toggle_wifi_state(self): |
| """Toggle WiFi state ON and OFF for N times.""" |
| for count in range(self.stress_count): |
| """Test toggling wifi""" |
| try: |
| self.log.debug("Going from on to off.") |
| wutils.wifi_toggle_state(self.dut, False) |
| self.log.debug("Going from off to on.") |
| startTime = time.time() |
| wutils.wifi_toggle_state(self.dut, True) |
| startup_time = time.time() - startTime |
| self.log.debug("WiFi was enabled on the device in %s s." % |
| startup_time) |
| except: |
| signals.TestFailure(details="", extras={"Iterations":"%d" % |
| self.stress_count, "Pass":"%d" %count}) |
| raise signals.TestPass(details="", extras={"Iterations":"%d" % |
| self.stress_count, "Pass":"%d" %(count+1)}) |
| |
| @test_tracker_info(uuid="49e3916a-9580-4bf7-a60d-a0f2545dcdde") |
| def test_stress_connect_traffic_disconnect_5g(self): |
| """Test to connect and disconnect from a network for N times. |
| |
| Steps: |
| 1. Scan and connect to a network. |
| 2. Run IPerf to upload data for few seconds. |
| 3. Disconnect. |
| 4. Repeat 1-3. |
| |
| """ |
| for count in range(self.stress_count): |
| try: |
| net_id = self.dut.droid.wifiAddNetwork(self.wpa_5g) |
| asserts.assert_true(net_id != -1, "Add network %r failed" % self.wpa_5g) |
| self.scan_and_connect_by_id(self.wpa_5g, net_id) |
| # Start IPerf traffic from phone to server. |
| # Upload data for 10s. |
| args = "-p {} -t {}".format(self.iperf_server.port, 10) |
| self.log.info("Running iperf client {}".format(args)) |
| result, data = self.dut.run_iperf_client(self.iperf_server_address, args) |
| if not result: |
| self.log.debug("Error occurred in iPerf traffic.") |
| self.run_ping(10) |
| wutils.wifi_forget_network(self.dut,self.wpa_5g[WifiEnums.SSID_KEY]) |
| time.sleep(WAIT_BEFORE_CONNECTION) |
| except: |
| raise signals.TestFailure("Network connect-disconnect failed." |
| "Look at logs", extras={"Iterations":"%d" % |
| self.stress_count, "Pass":"%d" %count}) |
| raise signals.TestPass(details="", extras={"Iterations":"%d" % |
| self.stress_count, "Pass":"%d" %(count+1)}) |
| |
| @test_tracker_info(uuid="e9827dff-0755-43ec-8b50-1f9756958460") |
| def test_stress_connect_long_traffic_5g(self): |
| """Test to connect to network and hold connection for few hours. |
| |
| Steps: |
| 1. Scan and connect to a network. |
| 2. Run IPerf to download data for few hours. |
| 3. Verify no WiFi disconnects/data interruption. |
| |
| """ |
| try: |
| self.scan_and_connect_by_ssid(self.wpa_5g) |
| # Start IPerf traffic from server to phone. |
| # Download data for 5 hours. |
| sec = self.stress_hours * 60 * 60 |
| args = "-p {} -t {} -R".format(self.iperf_server.port, sec) |
| self.log.info("Running iperf client {}".format(args)) |
| result, data = self.dut.run_iperf_client(self.iperf_server_address, |
| args, timeout=sec+1) |
| if not result: |
| self.log.debug("Error occurred in iPerf traffic.") |
| self.run_ping(sec) |
| except: |
| raise signals.TestFailure("Network long-connect failed." |
| "Look at logs", extras={"Total Hours":"%d" %self.stress_hours, |
| "Seconds Run":"UNKNOWN"}) |
| raise signals.TestPass(details="", extras={"Total Hours":"%d" % |
| self.stress_hours, "Seconds":"%d" %sec}) |
| |
| @test_tracker_info(uuid="d367c83e-5b00-4028-9ed8-f7b875997d13") |
| def test_stress_wifi_failover(self): |
| """This test does aggressive failover to several networks in list. |
| |
| Steps: |
| 1. Add and enable few networks. |
| 2. Let device auto-connect. |
| 3. Remove the connected network. |
| 4. Repeat 2-3. |
| 5. Device should connect to a network until all networks are |
| exhausted. |
| |
| """ |
| for count in range(int(self.stress_count/4)): |
| ssids = list() |
| for network in self.networks: |
| ssids.append(network[WifiEnums.SSID_KEY]) |
| ret = self.dut.droid.wifiAddNetwork(network) |
| asserts.assert_true(ret != -1, "Add network %r failed" % network) |
| self.dut.droid.wifiEnableNetwork(ret, 0) |
| time.sleep(WAIT_FOR_AUTO_CONNECT) |
| cur_network = self.dut.droid.wifiGetConnectionInfo() |
| cur_ssid = cur_network[WifiEnums.SSID_KEY] |
| self.log.info("Cur_ssid = %s" % cur_ssid) |
| for i in range(0,len(self.networks)): |
| self.log.debug("Forget network %s" % cur_ssid) |
| wutils.wifi_forget_network(self.dut, cur_ssid) |
| time.sleep(WAIT_FOR_AUTO_CONNECT) |
| cur_network = self.dut.droid.wifiGetConnectionInfo() |
| cur_ssid = cur_network[WifiEnums.SSID_KEY] |
| self.log.info("Cur_ssid = %s" % cur_ssid) |
| if i == len(self.networks) - 1: |
| break |
| if cur_ssid not in ssids: |
| raise signals.TestFailure("Device did not failover to the " |
| "expected network. SSID = %s" % cur_ssid) |
| network_config = self.dut.droid.wifiGetConfiguredNetworks() |
| self.log.info("Network Config = %s" % network_config) |
| if len(network_config): |
| raise signals.TestFailure("All the network configurations were not " |
| "removed. Configured networks = %s" % network_config, |
| extras={"Iterations":"%d" % self.stress_count, |
| "Pass":"%d" %(count*4)}) |
| raise signals.TestPass(details="", extras={"Iterations":"%d" % |
| self.stress_count, "Pass":"%d" %((count+1)*4)}) |
| |
| @test_tracker_info(uuid="2c19e8d1-ac16-4d7e-b309-795144e6b956") |
| def test_stress_softAP_startup_and_stop_5g(self): |
| """Test to bring up softAP and down for N times. |
| |
| Steps: |
| 1. Bring up softAP on 5G. |
| 2. Check for softAP on teh client device. |
| 3. Turn ON WiFi. |
| 4. Verify softAP is turned down and WiFi is up. |
| |
| """ |
| # Set country code explicitly to "US". |
| self.dut.droid.wifiSetCountryCode(wutils.WifiEnums.CountryCode.US) |
| self.dut_client.droid.wifiSetCountryCode(wutils.WifiEnums.CountryCode.US) |
| ap_ssid = "softap_" + utils.rand_ascii_str(8) |
| ap_password = utils.rand_ascii_str(8) |
| self.dut.log.info("softap setup: %s %s", ap_ssid, ap_password) |
| config = {wutils.WifiEnums.SSID_KEY: ap_ssid} |
| config[wutils.WifiEnums.PWD_KEY] = ap_password |
| for count in range(self.stress_count): |
| initial_wifi_state = self.dut.droid.wifiCheckState() |
| wutils.start_wifi_tethering(self.dut, |
| ap_ssid, |
| ap_password, |
| WifiEnums.WIFI_CONFIG_APBAND_5G) |
| wutils.start_wifi_connection_scan_and_ensure_network_found( |
| self.dut_client, ap_ssid) |
| wutils.stop_wifi_tethering(self.dut) |
| asserts.assert_false(self.dut.droid.wifiIsApEnabled(), |
| "SoftAp failed to shutdown!") |
| time.sleep(TIMEOUT) |
| cur_wifi_state = self.dut.droid.wifiCheckState() |
| if initial_wifi_state != cur_wifi_state: |
| raise signals.TestFailure("Wifi state was %d before softAP and %d now!" % |
| (initial_wifi_state, cur_wifi_state), |
| extras={"Iterations":"%d" % self.stress_count, |
| "Pass":"%d" %count}) |
| raise signals.TestPass(details="", extras={"Iterations":"%d" % |
| self.stress_count, "Pass":"%d" %(count+1)}) |
| |
| @test_tracker_info(uuid="eb22e26b-95d1-4580-8c76-85dfe6a42a0f") |
| def test_stress_wifi_roaming(self): |
| AP1_network = self.reference_networks[0]["5g"] |
| AP2_network = self.reference_networks[1]["5g"] |
| wutils.set_attns(self.attenuators, "AP1_on_AP2_off") |
| self.scan_and_connect_by_ssid(AP1_network) |
| # Reduce iteration to half because each iteration does two roams. |
| for count in range(int(self.stress_count/2)): |
| self.log.info("Roaming iteration %d, from %s to %s", count, |
| AP1_network, AP2_network) |
| try: |
| wutils.trigger_roaming_and_validate(self.dut, self.attenuators, |
| "AP1_off_AP2_on", AP2_network) |
| self.log.info("Roaming iteration %d, from %s to %s", count, |
| AP2_network, AP1_network) |
| wutils.trigger_roaming_and_validate(self.dut, self.attenuators, |
| "AP1_on_AP2_off", AP1_network) |
| except: |
| raise signals.TestFailure("Roaming failed. Look at logs", |
| extras={"Iterations":"%d" %self.stress_count, "Pass":"%d" % |
| (count*2)}) |
| raise signals.TestPass(details="", extras={"Iterations":"%d" % |
| self.stress_count, "Pass":"%d" %((count+1)*2)}) |
| |