1#!/usr/bin/env python3 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 itertools 18import pprint 19import time 20 21import acts.signals 22import acts.test_utils.wifi.wifi_test_utils as wutils 23 24from acts import asserts 25from acts.test_decorators import test_tracker_info 26from acts.test_utils.wifi.WifiBaseTest import WifiBaseTest 27from acts.controllers import iperf_server as ipf 28 29import json 30import logging 31import math 32import os 33from acts import utils 34import csv 35 36WifiEnums = wutils.WifiEnums 37 38 39class WifiIOTTwPkg1Test(WifiBaseTest): 40 """ Tests for wifi IOT 41 42 Test Bed Requirement: 43 * One Android device 44 * Wi-Fi IOT networks visible to the device 45 """ 46 47 def setup_class(self): 48 super().setup_class() 49 50 self.dut = self.android_devices[0] 51 wutils.wifi_test_device_init(self.dut) 52 53 req_params = [ "iot_networks", ] 54 opt_params = [ "open_network", 55 "iperf_server_address","iperf_port_arg", 56 "pdu_address" , "pduon_wait_time","pduon_address" 57 ] 58 self.unpack_userparams(req_param_names=req_params, 59 opt_param_names=opt_params) 60 61 asserts.assert_true( 62 len(self.iot_networks) > 0, 63 "Need at least one iot network with psk.") 64 65 if getattr(self, 'open_network', False): 66 self.iot_networks.append(self.open_network) 67 68 wutils.wifi_toggle_state(self.dut, True) 69 if "iperf_server_address" in self.user_params: 70 self.iperf_server = self.iperf_servers[0] 71 72 # create hashmap for testcase name and SSIDs 73 self.iot_test_prefix = "test_iot_connection_to_" 74 self.ssid_map = {} 75 for network in self.iot_networks: 76 SSID = network['SSID'].replace('-','_') 77 self.ssid_map[SSID] = network 78 79 # create folder for IOT test result 80 self.log_path = os.path.join(logging.log_path, "IOT_results") 81 os.makedirs(self.log_path, exist_ok=True) 82 83 Header=("test_name","throughput_TX","throughput_RX") 84 self.csv_write(Header) 85 86 # check pdu_address 87 if "pdu_address" and "pduon_wait_time" in self.user_params: 88 self.pdu_func() 89 90 def setup_test(self): 91 self.dut.droid.wakeLockAcquireBright() 92 self.dut.droid.wakeUpNow() 93 94 def teardown_test(self): 95 self.dut.droid.wakeLockRelease() 96 self.dut.droid.goToSleepNow() 97 wutils.reset_wifi(self.dut) 98 99 def teardown_class(self): 100 if "iperf_server_address" in self.user_params: 101 self.iperf_server.stop() 102 103 def on_fail(self, test_name, begin_time): 104 self.dut.take_bug_report(test_name, begin_time) 105 self.dut.cat_adb_log(test_name, begin_time) 106 107 """Helper Functions""" 108 109 def connect_to_wifi_network(self, network): 110 """Connection logic for open and psk wifi networks. 111 112 Args: 113 params: Dictionary with network info. 114 """ 115 SSID = network[WifiEnums.SSID_KEY] 116 self.dut.ed.clear_all_events() 117 wutils.start_wifi_connection_scan(self.dut) 118 scan_results = self.dut.droid.wifiGetScanResults() 119 wutils.assert_network_in_list({WifiEnums.SSID_KEY: SSID}, scan_results) 120 wutils.wifi_connect(self.dut, network, num_of_tries=3) 121 122 def run_iperf_client(self, network): 123 """Run iperf TX throughput after connection. 124 125 Args: 126 params: Dictionary with network info. 127 """ 128 if "iperf_server_address" in self.user_params: 129 130 # Add iot_result 131 iot_result = [] 132 self.iperf_server.start(tag="TX_server_{}".format( 133 self.current_test_name)) 134 wait_time = 5 135 SSID = network[WifiEnums.SSID_KEY] 136 self.log.info("Starting iperf traffic TX through {}".format(SSID)) 137 time.sleep(wait_time) 138 port_arg = "-p {} -J {}".format(self.iperf_server.port,self.iperf_port_arg) 139 success, data = self.dut.run_iperf_client(self.iperf_server_address, 140 port_arg) 141 # Parse and log result 142 client_output_path = os.path.join( 143 self.iperf_server.log_path, "IperfDUT,{},TX_client_{}".format( 144 self.iperf_server.port,self.current_test_name)) 145 with open(client_output_path, 'w') as out_file: 146 out_file.write("\n".join(data)) 147 self.iperf_server.stop() 148 149 iperf_file = self.iperf_server.log_files[-1] 150 try: 151 iperf_result = ipf.IPerfResult(iperf_file) 152 curr_throughput = math.fsum(iperf_result.instantaneous_rates) 153 except: 154 self.log.warning( 155 "ValueError: Cannot get iperf result. Setting to 0") 156 curr_throughput = 0 157 iot_result.append(curr_throughput) 158 self.log.info("Throughput is {0:.2f} Mbps".format(curr_throughput)) 159 160 self.log.debug(pprint.pformat(data)) 161 asserts.assert_true(success, "Error occurred in iPerf traffic.") 162 return iot_result 163 164 def run_iperf_server(self, network): 165 """Run iperf RX throughput after connection. 166 167 Args: 168 params: Dictionary with network info. 169 170 Returns: 171 iot_result: dict containing iot_results 172 """ 173 if "iperf_server_address" in self.user_params: 174 175 # Add iot_result 176 iot_result = [] 177 self.iperf_server.start(tag="RX_client_{}".format( 178 self.current_test_name)) 179 wait_time = 5 180 SSID = network[WifiEnums.SSID_KEY] 181 self.log.info("Starting iperf traffic RX through {}".format(SSID)) 182 time.sleep(wait_time) 183 port_arg = "-p {} -J -R {}".format(self.iperf_server.port,self.iperf_port_arg) 184 success, data = self.dut.run_iperf_client(self.iperf_server_address, 185 port_arg) 186 client_output_path = os.path.join( 187 self.iperf_server.log_path, "IperfDUT,{},RX_server_{}".format( 188 self.iperf_server.port,self.current_test_name)) 189 with open(client_output_path, 'w') as out_file: 190 out_file.write("\n".join(data)) 191 self.iperf_server.stop() 192 193 iperf_file = client_output_path 194 try: 195 iperf_result = ipf.IPerfResult(iperf_file) 196 curr_throughput = math.fsum(iperf_result.instantaneous_rates) 197 except: 198 self.log.warning( 199 "ValueError: Cannot get iperf result. Setting to 0") 200 curr_throughput = 0 201 iot_result.append(curr_throughput) 202 self.log.info("Throughput is {0:.2f} Mbps".format(curr_throughput)) 203 204 self.log.debug(pprint.pformat(data)) 205 asserts.assert_true(success, "Error occurred in iPerf traffic.") 206 return iot_result 207 208 def iperf_test_func(self,network): 209 """Main function to test iperf TX/RX. 210 211 Args: 212 params: Dictionary with network info 213 """ 214 # Initialize 215 iot_result = {} 216 217 # Run RvR and log result 218 iot_result["throughput_TX"] = self.run_iperf_client(network) 219 iot_result["throughput_RX"] = self.run_iperf_server(network) 220 iot_result["test_name"] = self.current_test_name 221 222 # Save output as text file 223 results_file_path = "{}/{}.json".format(self.log_path, 224 self.current_test_name) 225 with open(results_file_path, 'w') as results_file: 226 json.dump(iot_result, results_file, indent=4) 227 228 data=(iot_result["test_name"],iot_result["throughput_TX"][0], 229 iot_result["throughput_RX"][0]) 230 self.csv_write(data) 231 232 def csv_write(self,data): 233 with open("{}/Result.csv".format(self.log_path), "a", newline="") as csv_file: 234 csv_writer = csv.writer(csv_file,delimiter=',') 235 csv_writer.writerow(data) 236 csv_file.close() 237 238 def pdu_func(self): 239 """control Power Distribution Units on local machine. 240 241 Logic steps are 242 1. Turn off PDU for all port. 243 2. Turn on PDU for specified port. 244 """ 245 out_file_name = "PDU.log" 246 self.full_out_path = os.path.join(self.log_path, out_file_name) 247 cmd = "curl http://snmp:1234@{}/offs.cgi?led=11111111> {}".format(self.pdu_address, 248 self.full_out_path) 249 self.pdu_process = utils.start_standing_subprocess(cmd) 250 wait_time = 10 251 self.log.info("Starting set PDU to OFF") 252 time.sleep(wait_time) 253 self.full_out_path = os.path.join(self.log_path, out_file_name) 254 cmd = "curl http://snmp:1234@{}/ons.cgi?led={}> {}".format(self.pdu_address, 255 self.pduon_address, 256 self.full_out_path) 257 self.pdu_process = utils.start_standing_subprocess(cmd) 258 wait_time = int("{}".format(self.pduon_wait_time)) 259 self.log.info("Starting set PDU to ON for port1," 260 "wait for {}s".format(self.pduon_wait_time)) 261 time.sleep(wait_time) 262 self.log.info("PDU setup complete") 263 264 def connect_to_wifi_network_and_run_iperf(self, network): 265 """Connection logic for open and psk wifi networks. 266 267 Logic steps are 268 1. Connect to the network. 269 2. Run iperf throghput. 270 271 Args: 272 params: A dictionary with network info. 273 """ 274 self.connect_to_wifi_network(network) 275 self.iperf_test_func(network) 276 277 """Tests""" 278 279 @test_tracker_info(uuid="0e4ad6ed-595c-4629-a4c9-c6be9c3c58e0") 280 def test_iot_connection_to_ASUS_RT_AC68U_2G(self): 281 ssid_key = self.current_test_name.replace(self.iot_test_prefix, "") 282 self.connect_to_wifi_network_and_run_iperf(self.ssid_map[ssid_key]) 283 284 @test_tracker_info(uuid="a76d8acc-808e-4a5d-a52b-5ba07d07b810") 285 def test_iot_connection_to_ASUS_RT_AC68U_5G(self): 286 ssid_key = self.current_test_name.replace(self.iot_test_prefix, "") 287 self.connect_to_wifi_network_and_run_iperf(self.ssid_map[ssid_key]) 288 289 @test_tracker_info(uuid="659a3e5e-07eb-4905-9cda-92e959c7b674") 290 def test_iot_connection_to_D_Link_DIR_868L_2G(self): 291 ssid_key = self.current_test_name.replace(self.iot_test_prefix, "") 292 self.connect_to_wifi_network_and_run_iperf(self.ssid_map[ssid_key]) 293 294 @test_tracker_info(uuid="6bcfd736-30fc-48a8-b4fb-723d1d113f3c") 295 def test_iot_connection_to_D_Link_DIR_868L_5G(self): 296 ssid_key = self.current_test_name.replace(self.iot_test_prefix, "") 297 self.connect_to_wifi_network_and_run_iperf(self.ssid_map[ssid_key]) 298 299 @test_tracker_info(uuid="c9da945a-2c4a-44e1-881d-adf307b39b21") 300 def test_iot_connection_to_TP_LINK_WR940N_2G(self): 301 ssid_key = self.current_test_name.replace(self.iot_test_prefix, "") 302 self.connect_to_wifi_network_and_run_iperf(self.ssid_map[ssid_key]) 303 304 @test_tracker_info(uuid="db0d224d-df81-401f-bf35-08ad02e41a71") 305 def test_iot_connection_to_ASUS_RT_N66U_2G(self): 306 ssid_key = self.current_test_name.replace(self.iot_test_prefix, "") 307 self.connect_to_wifi_network_and_run_iperf(self.ssid_map[ssid_key]) 308 309 @test_tracker_info(uuid="845ff1d6-618d-40f3-81c3-6ed3a0751fde") 310 def test_iot_connection_to_ASUS_RT_N66U_5G(self): 311 ssid_key = self.current_test_name.replace(self.iot_test_prefix, "") 312 self.connect_to_wifi_network_and_run_iperf(self.ssid_map[ssid_key]) 313 314 @test_tracker_info(uuid="6908039b-ccc9-4777-a0f1-3494ce642014") 315 def test_iot_connection_to_ASUS_RT_AC54U_2G(self): 316 ssid_key = self.current_test_name.replace(self.iot_test_prefix, "") 317 self.connect_to_wifi_network_and_run_iperf(self.ssid_map[ssid_key]) 318 319 @test_tracker_info(uuid="2647c15f-2aad-47d7-8dee-b2ee1ac4cef6") 320 def test_iot_connection_to_ASUS_RT_AC54U_5G(self): 321 ssid_key = self.current_test_name.replace(self.iot_test_prefix, "") 322 self.connect_to_wifi_network_and_run_iperf(self.ssid_map[ssid_key]) 323 324 @test_tracker_info(uuid="99678f66-ddf1-454d-87e4-e55177ec380d") 325 def test_iot_connection_to_ASUS_RT_N56U_2G(self): 326 ssid_key = self.current_test_name.replace(self.iot_test_prefix, "") 327 self.connect_to_wifi_network_and_run_iperf(self.ssid_map[ssid_key]) 328 329 @test_tracker_info(uuid="4dd75e81-9a8e-44fd-9449-09f5ab8a63c3") 330 def test_iot_connection_to_ASUS_RT_N56U_5G(self): 331 ssid_key = self.current_test_name.replace(self.iot_test_prefix, "") 332 self.connect_to_wifi_network_and_run_iperf(self.ssid_map[ssid_key]) 333 334 @test_tracker_info(uuid="315397ce-50d5-4abf-a11c-1abcaef832d3") 335 def test_iot_connection_to_BELKIN_F9K1002v1_2G(self): 336 ssid_key = self.current_test_name.replace(self.iot_test_prefix, "") 337 self.connect_to_wifi_network_and_run_iperf(self.ssid_map[ssid_key]) 338 339 @test_tracker_info(uuid="05ba464a-b1ef-4ac1-a32f-c919ec4aa1dd") 340 def test_iot_connection_to_CISCO_E1200_2G(self): 341 ssid_key = self.current_test_name.replace(self.iot_test_prefix, "") 342 self.connect_to_wifi_network_and_run_iperf(self.ssid_map[ssid_key]) 343 344 @test_tracker_info(uuid="04912868-4a47-40ce-877e-4e4c89849557") 345 def test_iot_connection_to_TP_LINK_C2_2G(self): 346 ssid_key = self.current_test_name.replace(self.iot_test_prefix, "") 347 self.connect_to_wifi_network_and_run_iperf(self.ssid_map[ssid_key]) 348 349 @test_tracker_info(uuid="53517a21-3802-4185-b8bb-6eaace063a42") 350 def test_iot_connection_to_TP_LINK_C2_5G(self): 351 ssid_key = self.current_test_name.replace(self.iot_test_prefix, "") 352 self.connect_to_wifi_network_and_run_iperf(self.ssid_map[ssid_key]) 353 354 @test_tracker_info(uuid="71c08c1c-415d-4da4-a151-feef43fb6ad8") 355 def test_iot_connection_to_ASUS_RT_AC66U_2G(self): 356 ssid_key = self.current_test_name.replace(self.iot_test_prefix, "") 357 self.connect_to_wifi_network_and_run_iperf(self.ssid_map[ssid_key]) 358 359 @test_tracker_info(uuid="2322c155-07d1-47c9-bd21-2e358e3df6ee") 360 def test_iot_connection_to_ASUS_RT_AC66U_5G(self): 361 ssid_key = self.current_test_name.replace(self.iot_test_prefix, "") 362 self.connect_to_wifi_network_and_run_iperf(self.ssid_map[ssid_key]) 363