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