#!/usr/bin/env python3 # # Copyright 2020 - 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 csv import json import os import time import acts_contrib.test_utils.wifi.wifi_test_utils as wutils from acts import signals from acts.test_decorators import test_tracker_info from acts_contrib.test_utils.wifi.WifiBaseTest import WifiBaseTest from acts_contrib.test_utils.wifi import pdu_controller_utils WifiEnums = wutils.WifiEnums WAIT_BETWEEN_ACTIONS = 5 class WifiIOTConnectionTest(WifiBaseTest): """ Tests for wifi IOT Connection Test Bed Requirement: * One Android device * Wi-Fi IOT networks visible to the device """ def __init__(self, controllers): WifiBaseTest.__init__(self, controllers) self.generate_test_list() def setup_class(self): self.dut = self.android_devices[0] self.pdu = pdu_controller_utils.create(self.user_params['Pdu'])[0] if hasattr(self, 'packet_capture'): self.packet_capture = self.packet_capture[0] wutils.wifi_test_device_init(self.dut) self.csv_write(["Project", "TimeStamp", "Build_ID", "Test_Name", "Host", "RSSI", "Link_speed", "freq", "#iteration", "#Pass"], ) req_params = ['iot_password', 'pdu_wait_time', 'iot_ssid', 'Pdu', 'iot_connection_iteration'] self.unpack_userparams( req_param_names=req_params) def setup_test(self): self.dut.droid.wakeLockAcquireBright() self.dut.droid.wakeUpNow() wutils.reset_wifi(self.dut) def teardown_test(self): self.dut.droid.wakeLockRelease() self.dut.droid.goToSleepNow() if hasattr(self, 'packet_capture'): wutils.stop_pcap(self.packet_capture, self.pcap_procs, False) 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 csv_write(self, data): """Output .CSV file as a result. Args: data: a dict containing: 'project', 'TimeStamp', 'test_name', 'test_name', 'host', 'rssi', 'link_speed', 'freq', '#Iteration', '#Pass' """ time_str = time.strftime("%Y-%m-%d") file_name_format = "wifi_iot_connection_" + time_str + ".csv" file_name = os.path.join(self.log_path, file_name_format) with open(file_name, "a", newline="") as csv_file: csv_writer = csv.writer(csv_file, delimiter=',') csv_writer.writerow(data) def get_wifi_info(self): """Get current connected WiFi AP's info. Returns: rssi: Wi-Fi rssi link_speed: Wi-Fi link speed freq: Wi-Fi freq """ freq = "NA" link_speed = "NA" rssi = "NA" try: out = self.dut.adb.shell("iw wlan0 link") if out: for line in out.split("\n"): if "freq:" in line: freq = line.split()[1] elif "signal" in line: rssi = line.split()[1] elif "bitrate" in line: link_speed = line.split()[2] except AttributeError as e: self.log.debug("No wifi info, check Wi-Fi connection.", e) finally: return rssi, link_speed, freq def getprop(self, property_name): """Get dut's property Args: property_name: property name, e.g., "ro.build.product" Return: property """ return self.dut.adb.getprop(property_name) def generate_test_list(self): """Generates a test list which is sorted by host number.""" sorted_list = sorted( self.user_params["iot_ssid"], key=lambda ssid: ssid["host"]) for test_item in sorted_list: self.init_test_case(self.wifi_iot_connection_test, test_item) def init_test_case(self, wifi_iot_connection_test, test_item): """Generates a single test case from the given data. Args: wifi_iot_connection_test: The base test case function to run. test_item: test case required info, include: "ssid", "uuid", "host", "band", "channel" """ test_name = test_item["ssid"] test_tracker_uuid = test_item["uuid"] if not test_name.startswith("test_"): test_name = "test_{}".format(test_name) test_case = test_tracker_info(uuid=test_tracker_uuid)( lambda: wifi_iot_connection_test(test_item, self.user_params["iot_password"])) setattr(self, test_name, test_case) self.tests.append(test_name) def pdu_status_check(self): """ Check pdu status if pdu's currently ON outlet != next test case's host number, switch ON outlet to the host number. for example: pdu's outlet 3 is on, next test case host number is 4, turn off pdu outlet 3 and turn on outlet 4. """ _, self.pdu_status = self.pdu.get_status() self.log.info("Current host of the pdu : {}".format(self.pdu_status)) self.log.info("Next host of the test case : {}".format(self.host)) if str(self.host) != self.pdu_status: self.pdu.off_all() self.log.info("Switch PDU to {}".format(self.host)) self.pdu.turn_on_outlets(str(self.host)) self.log.info("Wait {} secs to bring up the APs" .format(self.user_params["pdu_wait_time"])) time.sleep(self.user_params["pdu_wait_time"]) def start_packet_capture(self, band, channel): """Configure wireless packet capture on pre-defined channels. Args: band: '2G' or '5G'. pre-defined in config file. channel: Wi-fi Channel, pre-defined in the config file. """ self.log.info("Capturing packets from Channel: {}".format(channel)) result = self.packet_capture.configure_monitor_mode(band, channel) if not result: self.dut.log.error("Failed to configure channel " "for {} band".format(band)) self.pcap_procs = wutils.start_pcap( self.packet_capture, band, self.current_test_name) time.sleep(3) def ping_public_gateway_ip(self): """Ping 8.8.8.8""" try: ping_result = self.dut.adb.shell('ping -w 5 8.8.8.8') if '0%' in ping_result: self.dut.log.info('Ping success') return True except: self.dut.log.error('Faild to ping public gateway 8.8.8.8') return False def connect_to_network_and_ping(self, network): ssid = network[WifiEnums.SSID_KEY] connection_pass = 0 for i in range(self.user_params['iot_connection_iteration']): self.dut.log.info('Connection iteration : {}'.format(i + 1)) try: wutils.connect_to_wifi_network(self.dut, network, num_of_connect_tries=1, check_connectivity=False) time.sleep(WAIT_BETWEEN_ACTIONS) self.rssi, self.link_speed, self.freq = self.get_wifi_info() time.sleep(WAIT_BETWEEN_ACTIONS) if self.ping_public_gateway_ip(): connection_pass += 1 wutils.wifi_forget_network(self.dut, ssid) self.dut.log.info("connection_pass: {}" .format(connection_pass)) time.sleep(WAIT_BETWEEN_ACTIONS) except: self.dut.log.error("connection_fail") # Create a dictionary to store data in a json file. connection_result = {} connection_result["project"] = self.getprop("ro.build.product") connection_result["TimeStamp"] = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) connection_result["Build ID"] = self.getprop("ro.build.id") connection_result["test_name"] = self.current_test_name connection_result["host"] = self.host connection_result['rssi'] = self.rssi connection_result['link_speed'] = self.link_speed connection_result['freq'] = self.freq connection_result['#Iteration'] = self.user_params[ 'iot_connection_iteration'] connection_result['#Pass'] = connection_pass # Create a json file for each test case. results_file_path = os.path.join(self.log_path, "{}.json".format( self.current_test_name)) with open(results_file_path, 'w') as results_file: json.dump(connection_result, results_file, indent=4) data = (self.getprop("ro.build.product"), time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()), self.getprop("ro.build.id"), connection_result["test_name"], connection_result["host"], connection_result['rssi'], connection_result['link_speed'], connection_result['freq'], connection_result['#Iteration'], connection_result['#Pass'] ) self.csv_write(data) self.rssi = "NA" self.link_speed = "NA" self.freq = "NA" if connection_pass < self.user_params['iot_connection_iteration']: raise signals.TestFailure("connection failed more than expected") def wifi_iot_connection_test(self, test_item, password): """The base test case logic for Wifi IOT generated test cases. 1. Check pdu outlets is ON as expected. 2. Scan SSIDs of 10 Wi-Fi APs. 3. Start Packet Capture on pre-defined channel. 3. Connect to the AP and ping Gateway 8.8.8.8 for 5 times. Args: test_item: Test info include: 'ssid', 'host', 'uuid', 'band', 'channel'. password: pwd for login to the access point. """ network = {'SSID': test_item["ssid"], 'password': password} self.host = test_item["host"] self.pdu_status_check() # Capturing wireless packets before DUT connect to Wi-Fi network. self.band = test_item["band"] self.channel = test_item["channel"] if hasattr(self, 'packet_capture'): self.start_packet_capture(self.band, self.channel) # Connect to Wi-Fi network and ping public gateway for 5 times. self.connect_to_network_and_ping(network)