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