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 |
Xianyuan Jia | 63751fb | 2020-11-17 00:07:40 +0000 | [diff] [blame^] | 22 | import acts_contrib.test_utils.wifi.wifi_test_utils as wutils |
Tim | 12e0114 | 2018-07-05 14:18:30 +0800 | [diff] [blame] | 23 | |
| 24 | from acts import asserts |
| 25 | from acts.test_decorators import test_tracker_info |
Xianyuan Jia | 63751fb | 2020-11-17 00:07:40 +0000 | [diff] [blame^] | 26 | from acts_contrib.test_utils.wifi.WifiBaseTest import WifiBaseTest |
Tim | 12e0114 | 2018-07-05 14:18:30 +0800 | [diff] [blame] | 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 | |
Tim | 12e0114 | 2018-07-05 14:18:30 +0800 | [diff] [blame] | 47 | def setup_class(self): |
Xianyuan Jia | 168103b | 2019-09-06 12:22:52 -0700 | [diff] [blame] | 48 | super().setup_class() |
| 49 | |
Tim | 12e0114 | 2018-07-05 14:18:30 +0800 | [diff] [blame] | 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") |
Mark De Ruyter | 72f8df9 | 2020-02-12 13:44:49 -0800 | [diff] [blame] | 81 | os.makedirs(self.log_path, exist_ok=True) |
Tim | 12e0114 | 2018-07-05 14:18:30 +0800 | [diff] [blame] | 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): |
Xianyuan Jia | 97a865e | 2020-11-10 02:09:40 +0000 | [diff] [blame] | 91 | super().setup_test() |
Tim | 12e0114 | 2018-07-05 14:18:30 +0800 | [diff] [blame] | 92 | self.dut.droid.wakeLockAcquireBright() |
| 93 | self.dut.droid.wakeUpNow() |
| 94 | |
| 95 | def teardown_test(self): |
Xianyuan Jia | 97a865e | 2020-11-10 02:09:40 +0000 | [diff] [blame] | 96 | super().teardown_test() |
Tim | 12e0114 | 2018-07-05 14:18:30 +0800 | [diff] [blame] | 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 | |
Tim | 12e0114 | 2018-07-05 14:18:30 +0800 | [diff] [blame] | 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]) |