1#!/usr/bin/env python3 2# 3# Copyright 2020 - 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 csv 18import json 19import os 20import time 21import acts_contrib.test_utils.wifi.wifi_test_utils as wutils 22from acts import signals 23from acts.test_decorators import test_tracker_info 24from acts_contrib.test_utils.wifi.WifiBaseTest import WifiBaseTest 25from acts_contrib.test_utils.wifi import pdu_controller_utils 26 27WifiEnums = wutils.WifiEnums 28WAIT_BETWEEN_ACTIONS = 5 29 30class WifiIOTConnectionTest(WifiBaseTest): 31 """ Tests for wifi IOT Connection 32 33 Test Bed Requirement: 34 * One Android device 35 * Wi-Fi IOT networks visible to the device 36 """ 37 38 def __init__(self, controllers): 39 WifiBaseTest.__init__(self, controllers) 40 self.generate_test_list() 41 42 def setup_class(self): 43 self.dut = self.android_devices[0] 44 self.pdu = pdu_controller_utils.create(self.user_params['Pdu'])[0] 45 46 if hasattr(self, 'packet_capture'): 47 self.packet_capture = self.packet_capture[0] 48 49 wutils.wifi_test_device_init(self.dut) 50 51 self.csv_write(["Project", "TimeStamp", "Build_ID", "Test_Name", 52 "Host", "RSSI", "Link_speed", "freq", "#iteration", 53 "#Pass"], ) 54 55 req_params = ['iot_password', 'pdu_wait_time', 56 'iot_ssid', 'Pdu', 'iot_connection_iteration'] 57 self.unpack_userparams( 58 req_param_names=req_params) 59 60 def setup_test(self): 61 self.dut.droid.wakeLockAcquireBright() 62 self.dut.droid.wakeUpNow() 63 wutils.reset_wifi(self.dut) 64 65 def teardown_test(self): 66 self.dut.droid.wakeLockRelease() 67 self.dut.droid.goToSleepNow() 68 wutils.stop_pcap(self.packet_capture, self.pcap_procs, False) 69 70 def on_fail(self, test_name, begin_time): 71 self.dut.take_bug_report(test_name, begin_time) 72 self.dut.cat_adb_log(test_name, begin_time) 73 74 def csv_write(self, data): 75 """Output .CSV file as a result. 76 77 Args: 78 data: a dict containing: 79 'project', 'TimeStamp', 'test_name', 'test_name', 80 'host', 'rssi', 'link_speed', 'freq', '#Iteration', '#Pass' 81 """ 82 time_str = time.strftime("%Y-%m-%d") 83 file_name_format = "wifi_iot_connection_" + time_str + ".csv" 84 file_name = os.path.join(self.log_path, file_name_format) 85 with open(file_name, "a", newline="") as csv_file: 86 csv_writer = csv.writer(csv_file, delimiter=',') 87 csv_writer.writerow(data) 88 89 def get_wifi_info(self): 90 """Get current connected WiFi AP's info. 91 92 Returns: 93 rssi: Wi-Fi rssi 94 link_speed: Wi-Fi link speed 95 freq: Wi-Fi freq 96 """ 97 freq = "NA" 98 link_speed = "NA" 99 rssi = "NA" 100 101 try: 102 out = self.dut.adb.shell("iw wlan0 link") 103 if out: 104 for line in out.split("\n"): 105 if "freq:" in line: 106 freq = line.split()[1] 107 elif "signal" in line: 108 rssi = line.split()[1] 109 elif "bitrate" in line: 110 link_speed = line.split()[2] 111 except AttributeError as e: 112 self.log.debug("No wifi info, check Wi-Fi connection.", e) 113 finally: 114 return rssi, link_speed, freq 115 116 def getprop(self, property_name): 117 """Get dut's property 118 119 Args: 120 property_name: property name, e.g., "ro.build.product" 121 Return: 122 property 123 """ 124 return self.dut.adb.getprop(property_name) 125 126 def generate_test_list(self): 127 """Generates a test list which is sorted by host number.""" 128 sorted_list = sorted( 129 self.user_params["iot_ssid"], key=lambda ssid: ssid["host"]) 130 for test_item in sorted_list: 131 self.init_test_case(self.wifi_iot_connection_test, test_item) 132 133 def init_test_case(self, wifi_iot_connection_test, test_item): 134 """Generates a single test case from the given data. 135 136 Args: 137 wifi_iot_connection_test: The base test case function to run. 138 test_item: test case required info, include: 139 "ssid", "uuid", "host", "band", "channel" 140 """ 141 test_name = test_item["ssid"] 142 test_tracker_uuid = test_item["uuid"] 143 if not test_name.startswith("test_"): 144 test_name = "test_{}".format(test_name) 145 test_case = test_tracker_info(uuid=test_tracker_uuid)( 146 lambda: wifi_iot_connection_test(test_item, 147 self.user_params["iot_password"])) 148 setattr(self, test_name, test_case) 149 self.tests.append(test_name) 150 151 def pdu_status_check(self): 152 """ Check pdu status 153 if pdu's currently ON outlet != next test case's host number, 154 switch ON outlet to the host number. 155 for example: pdu's outlet 3 is on, next test case host number is 4, 156 turn off pdu outlet 3 and turn on outlet 4. 157 """ 158 _, self.pdu_status = self.pdu.get_status() 159 self.log.info("Current host of the pdu : {}".format(self.pdu_status)) 160 self.log.info("Next host of the test case : {}".format(self.host)) 161 if str(self.host) != self.pdu_status: 162 self.pdu.off_all() 163 self.log.info("Switch PDU to {}".format(self.host)) 164 self.pdu.turn_on_outlets(str(self.host)) 165 self.log.info("Wait {} secs to bring up the APs" 166 .format(self.user_params["pdu_wait_time"])) 167 time.sleep(self.user_params["pdu_wait_time"]) 168 169 def start_packet_capture(self, band, channel): 170 """Configure wireless packet capture on pre-defined channels. 171 172 Args: 173 band: '2G' or '5G'. pre-defined in config file. 174 channel: Wi-fi Channel, pre-defined in the config file. 175 """ 176 self.log.info("Capturing packets from Channel: {}".format(channel)) 177 result = self.packet_capture.configure_monitor_mode(band, channel) 178 if not result: 179 self.dut.log.error("Failed to configure channel " 180 "for {} band".format(band)) 181 self.pcap_procs = wutils.start_pcap( 182 self.packet_capture, band, self.current_test_name) 183 time.sleep(3) 184 185 def ping_public_gateway_ip(self): 186 """Ping 8.8.8.8""" 187 try: 188 ping_result = self.dut.adb.shell('ping -w 5 8.8.8.8') 189 if '0%' in ping_result: 190 self.dut.log.info('Ping success') 191 return True 192 except: 193 self.dut.log.error('Faild to ping public gateway 8.8.8.8') 194 return False 195 196 def connect_to_network_and_ping(self, network): 197 ssid = network[WifiEnums.SSID_KEY] 198 connection_pass = 0 199 for i in range(self.user_params['iot_connection_iteration']): 200 self.dut.log.info('Connection iteration : {}'.format(i + 1)) 201 try: 202 wutils.connect_to_wifi_network(self.dut, network, 203 num_of_connect_tries=1, 204 check_connectivity=False) 205 time.sleep(WAIT_BETWEEN_ACTIONS) 206 self.rssi, self.link_speed, self.freq = self.get_wifi_info() 207 time.sleep(WAIT_BETWEEN_ACTIONS) 208 if self.ping_public_gateway_ip(): 209 connection_pass += 1 210 wutils.wifi_forget_network(self.dut, ssid) 211 self.dut.log.info("connection_pass: {}" 212 .format(connection_pass)) 213 time.sleep(WAIT_BETWEEN_ACTIONS) 214 except: 215 self.dut.log.error("connection_fail") 216 217 # Create a dictionary to store data in a json file. 218 connection_result = {} 219 connection_result["project"] = self.getprop("ro.build.product") 220 connection_result["TimeStamp"] = time.strftime("%Y-%m-%d %H:%M:%S", 221 time.localtime()) 222 connection_result["Build ID"] = self.getprop("ro.build.id") 223 connection_result["test_name"] = self.current_test_name 224 connection_result["host"] = self.host 225 connection_result['rssi'] = self.rssi 226 connection_result['link_speed'] = self.link_speed 227 connection_result['freq'] = self.freq 228 connection_result['#Iteration'] = self.user_params[ 229 'iot_connection_iteration'] 230 connection_result['#Pass'] = connection_pass 231 232 # Create a json file for each test case. 233 results_file_path = os.path.join(self.log_path, "{}.json".format( 234 self.current_test_name)) 235 236 with open(results_file_path, 'w') as results_file: 237 json.dump(connection_result, results_file, indent=4) 238 239 data = (self.getprop("ro.build.product"), 240 time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()), 241 self.getprop("ro.build.id"), 242 connection_result["test_name"], 243 connection_result["host"], 244 connection_result['rssi'], 245 connection_result['link_speed'], 246 connection_result['freq'], 247 connection_result['#Iteration'], 248 connection_result['#Pass'] 249 ) 250 251 self.csv_write(data) 252 253 self.rssi = "NA" 254 self.link_speed = "NA" 255 self.freq = "NA" 256 257 if connection_pass < self.user_params['iot_connection_iteration']: 258 raise signals.TestFailure("connection failed more than expected") 259 260 def wifi_iot_connection_test(self, test_item, password): 261 """The base test case logic for Wifi IOT generated test cases. 262 263 1. Check pdu outlets is ON as expected. 264 2. Scan SSIDs of 10 Wi-Fi APs. 265 3. Start Packet Capture on pre-defined channel. 266 3. Connect to the AP and ping Gateway 8.8.8.8 for 5 times. 267 268 Args: 269 test_item: Test info include: 270 'ssid', 'host', 'uuid', 'band', 'channel'. 271 password: pwd for login to the access point. 272 """ 273 network = {'SSID': test_item["ssid"], 'password': password} 274 self.host = test_item["host"] 275 self.pdu_status_check() 276 277 # Capturing wireless packets before DUT connect to Wi-Fi network. 278 self.band = test_item["band"] 279 self.channel = test_item["channel"] 280 if hasattr(self, 'packet_capture'): 281 self.start_packet_capture(self.band, self.channel) 282 283 # Connect to Wi-Fi network and ping public gateway for 5 times. 284 self.connect_to_network_and_ping(network) 285