#!/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 re import random import time import acts.controllers.packet_capture as packet_capture import acts.signals as signals import acts_contrib.test_utils.wifi.rpm_controller_utils as rutils import acts_contrib.test_utils.wifi.wifi_datastore_utils as dutils import acts_contrib.test_utils.wifi.wifi_test_utils as wutils from acts import asserts from acts.base_test import BaseTestClass from acts.controllers.ap_lib import hostapd_constants from acts.test_decorators import test_tracker_info from acts_contrib.test_utils.wifi.WifiBaseTest import WifiBaseTest WifiEnums = wutils.WifiEnums WAIT_BEFORE_CONNECTION = 1 SINGLE_BAND = 1 DUAL_BAND = 2 TIMEOUT = 60 TEST = 'test_' PING_ADDR = 'www.google.com' NUM_LINK_PROBES = 3 PROBE_DELAY_SEC = 1 class WifiChaosTest(WifiBaseTest): """ Tests for wifi IOT Test Bed Requirement: * One Android device * Wi-Fi IOT networks visible to the device """ def __init__(self, configs): BaseTestClass.__init__(self, configs) self.generate_interop_tests() def randomize_testcases(self): """Randomize the list of hosts and build a random order of tests, based on SSIDs, keeping all the relevant bands of an AP together. """ temp_tests = list() hosts = self.user_params['interop_host'] random.shuffle(hosts) for host in hosts: ssid_2g = None ssid_5g = None info = dutils.show_device(host) # Based on the information in datastore, add to test list if # AP has 2G band. if 'ssid_2g' in info: ssid_2g = info['ssid_2g'] temp_tests.append(TEST + ssid_2g) # Based on the information in datastore, add to test list if # AP has 5G band. if 'ssid_5g' in info: ssid_5g = info['ssid_5g'] temp_tests.append(TEST + ssid_5g) self.tests = temp_tests def generate_interop_testcase(self, base_test, testcase_name, ssid_dict): """Generates a single test case from the given data. Args: base_test: The base test case function to run. testcase_name: The name of the test case. ssid_dict: The information about the network under test. """ ssid = testcase_name test_tracker_uuid = ssid_dict[testcase_name]['uuid'] hostname = ssid_dict[testcase_name]['host'] if not testcase_name.startswith('test_'): testcase_name = 'test_%s' % testcase_name test_case = test_tracker_info( uuid=test_tracker_uuid)(lambda: base_test(ssid, hostname)) setattr(self, testcase_name, test_case) self.tests.append(testcase_name) def generate_interop_tests(self): for ssid_dict in self.user_params['interop_ssid']: testcase_name = list(ssid_dict)[0] self.generate_interop_testcase(self.interop_base_test, testcase_name, ssid_dict) self.randomize_testcases() def setup_class(self): super().setup_class() self.dut = self.android_devices[0] self.admin = 'admin' + str(random.randint(1000001, 12345678)) wutils.wifi_test_device_init(self.dut) # Set country code explicitly to "US". wutils.set_wifi_country_code(self.dut, wutils.WifiEnums.CountryCode.US) asserts.assert_true( self.lock_pcap(), "Could not lock a Packet Capture. Aborting Interop test.") wutils.wifi_toggle_state(self.dut, True) def lock_pcap(self): """Lock a Packet Capturere to use for the test.""" # Get list of devices from the datastore. locked_pcap = False devices = dutils.get_devices() for device in devices: device_name = device['hostname'] device_type = device['ap_label'] if device_type == 'PCAP' and not device['lock_status']: if dutils.lock_device(device_name, self.admin): self.pcap_host = device_name host = device['ip_address'] self.log.info("Locked Packet Capture device: %s" % device_name) locked_pcap = True break else: self.log.warning("Failed to lock %s PCAP." % device_name) if not locked_pcap: return False pcap_config = {'ssh_config': {'user': 'root'}} pcap_config['ssh_config']['host'] = host self.pcap = packet_capture.PacketCapture(pcap_config) return True def setup_test(self): super().setup_test() self.dut.droid.wakeLockAcquireBright() self.dut.droid.wakeUpNow() def on_pass(self, test_name, begin_time): wutils.stop_pcap(self.pcap, self.pcap_procs, True) def on_fail(self, test_name, begin_time): # Sleep to ensure all failed packets are captured. time.sleep(5) wutils.stop_pcap(self.pcap, self.pcap_procs, False) super().on_fail(test_name, begin_time) def teardown_test(self): super().teardown_test() self.dut.droid.wakeLockRelease() self.dut.droid.goToSleepNow() def teardown_class(self): # Unlock the PCAP. if not dutils.unlock_device(self.pcap_host): self.log.warning("Failed to unlock %s PCAP. Check in datastore.") """Helper Functions""" 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("Finding Gateway...") route_response = self.dut.adb.shell("ip route get 8.8.8.8") gateway_ip = re.search('via (.*) dev', str(route_response)).group(1) self.log.info("Gateway IP = %s" % gateway_ip) self.log.info("Running ping for %d seconds" % sec) result = self.dut.adb.shell("ping -w %d %s" % (sec, gateway_ip), timeout=sec + 1) self.log.debug("Ping Result = %s" % result) if "100% packet loss" in result: raise signals.TestFailure("100% packet loss during ping") def send_link_probes(self, network): """ Send link probes, and verify that the device and AP did not crash. Also verify that at least one link probe succeeded. Steps: 1. Send a few link probes. 2. Ensure that the device and AP did not crash (by checking that the device remains connected to the expected network). """ results = wutils.send_link_probes(self.dut, NUM_LINK_PROBES, PROBE_DELAY_SEC) self.log.info("Link Probe results: %s" % (results, )) wifi_info = self.dut.droid.wifiGetConnectionInfo() expected = network[WifiEnums.SSID_KEY] actual = wifi_info[WifiEnums.SSID_KEY] asserts.assert_equal( expected, actual, "Device did not remain connected after sending link probes!") def unlock_and_turn_off_ap(self, hostname, rpm_port, rpm_ip): """UNlock the AP in datastore and turn off the AP. Args: hostname: Hostname of the AP. rpm_port: Port number on the RPM for the AP. rpm_ip: RPM's IP address. """ # Un-Lock AP in datastore. self.log.debug("Un-lock AP in datastore") if not dutils.unlock_device(hostname): self.log.warning("Failed to unlock %s AP. Check AP in datastore.") # Turn OFF AP from the RPM port. rutils.turn_off_ap(rpm_port, rpm_ip) def run_connect_disconnect(self, network, hostname, rpm_port, rpm_ip, release_ap): """Run connect/disconnect to a given network in loop. Args: network: Dict, network information. hostname: Hostanme of the AP to connect to. rpm_port: Port number on the RPM for the AP. rpm_ip: Port number on the RPM for the AP. release_ap: Flag to determine if we should turn off the AP yet. Raises: TestFailure if the network connection fails. """ for attempt in range(5): try: begin_time = time.time() ssid = network[WifiEnums.SSID_KEY] net_id = self.dut.droid.wifiAddNetwork(network) asserts.assert_true(net_id != -1, "Add network %s failed" % network) self.log.info("Connecting to %s" % ssid) self.scan_and_connect_by_id(network, net_id) self.run_ping(10) # TODO(b/133369482): uncomment once bug is resolved # self.send_link_probes(network) wutils.wifi_forget_network(self.dut, ssid) time.sleep(WAIT_BEFORE_CONNECTION) except Exception as e: self.log.error("Connection to %s network failed on the %d " "attempt with exception %s." % (ssid, attempt, e)) # TODO:(bmahadev) Uncomment after scan issue is fixed. # self.dut.take_bug_report(ssid, begin_time) # self.dut.cat_adb_log(ssid, begin_time) if release_ap: self.unlock_and_turn_off_ap(hostname, rpm_port, rpm_ip) raise signals.TestFailure("Failed to connect to %s" % ssid) def get_band_and_chan(self, ssid): """Get the band and channel information from SSID. Args: ssid: SSID of the network. """ ssid_info = ssid.split('_') self.band = ssid_info[-1] for item in ssid_info: # Skip over the router model part. if 'ch' in item and item != ssid_info[0]: self.chan = re.search(r'(\d+)', item).group(0) return raise signals.TestFailure("Channel information not found in SSID.") def interop_base_test(self, ssid, hostname): """Base test for all the connect-disconnect interop tests. Args: ssid: string, SSID of the network to connect to. hostname: string, hostname of the AP. Steps: 1. Lock AP in datstore. 2. Turn on AP on the rpm switch. 3. Run connect-disconnect in loop. 4. Turn off AP on the rpm switch. 5. Unlock AP in datastore. """ network = {} network['password'] = 'password' network['SSID'] = ssid release_ap = False wutils.reset_wifi(self.dut) # Lock AP in datastore. self.log.info("Lock AP in datastore") ap_info = dutils.show_device(hostname) # If AP is locked by a different test admin, then we skip. if ap_info['lock_status'] and ap_info['locked_by'] != self.admin: raise signals.TestSkip("AP %s is locked, skipping test" % hostname) if not dutils.lock_device(hostname, self.admin): self.log.warning("Failed to lock %s AP. Unlock AP in datastore" " and try again.") raise signals.TestFailure("Failed to lock AP") band = SINGLE_BAND if ('ssid_2g' in ap_info) and ('ssid_5g' in ap_info): band = DUAL_BAND if (band == SINGLE_BAND) or (band == DUAL_BAND and '5G' in ssid): release_ap = True # Get AP RPM attributes and Turn ON AP. rpm_ip = ap_info['rpm_ip'] rpm_port = ap_info['rpm_port'] rutils.turn_on_ap(self.pcap, ssid, rpm_port, rpm_ip=rpm_ip) self.log.info("Finished turning ON AP.") # Experimental. Some APs take upto a min to come online. time.sleep(60) self.get_band_and_chan(ssid) self.pcap.configure_monitor_mode(self.band, self.chan) self.pcap_procs = wutils.start_pcap(self.pcap, self.band.lower(), self.test_name) self.run_connect_disconnect(network, hostname, rpm_port, rpm_ip, release_ap) # Un-lock only if it's a single band AP or we are running the last band. if release_ap: self.unlock_and_turn_off_ap(hostname, rpm_port, rpm_ip)