1# 2# Copyright 2019 - The Android Open Source Project 3# 4# Licensed under the Apache License, Version 2.0 (the "License"); 5# you may not use this file except in compliance with the License. 6# You may obtain a copy of the License at 7# 8# http://www.apache.org/licenses/LICENSE-2.0 9# 10# Unless required by applicable law or agreed to in writing, software 11# distributed under the License is distributed on an "AS IS" BASIS, 12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 13# See the License for the specific language governing permissions and 14# limitations under the License. 15 16import math 17import os 18 19import time 20 21import threading 22from acts import utils 23from acts import signals 24from acts import asserts 25from acts.controllers import attenuator 26from acts.controllers.sl4a_lib import rpc_client 27from acts.test_decorators import test_tracker_info 28from acts_contrib.test_utils.net.net_test_utils import start_tcpdump, stop_tcpdump 29from acts_contrib.test_utils.wifi import wifi_test_utils as wutils 30from acts_contrib.test_utils.wifi.wifi_test_utils import WifiEnums 31from acts_contrib.test_utils.wifi.WifiBaseTest import WifiBaseTest 32from acts.utils import stop_standing_subprocess 33 34TCPDUMP_PATH = '/data/local/tmp/tcpdump' 35 36 37class WifiIFSTwTest(WifiBaseTest): 38 """Tests for wifi IFS 39 40 Test Bed Requirement: 41 *One Android device 42 *Two Visible Wi-Fi Access Points 43 *One attenuator with 4 ports 44 """ 45 46 def setup_class(self): 47 """Setup required dependencies from config file and configure 48 the required networks for testing roaming. 49 50 Returns: 51 True if successfully configured the requirements for testing. 52 """ 53 super().setup_class() 54 self.simulation_thread_running = False 55 self.atten_roaming_count = 0 56 self.start_db = 30 57 self.roaming_cycle_seconds = 20 58 self.fail_count = 0 59 self.retry_pass_count = 0 60 self.ping_count = 0 61 62 self.dut = self.android_devices[0] 63 wutils.wifi_test_device_init(self.dut) 64 req_params = ["attenuators", "ifs_params"] 65 opt_param = [] 66 67 self.unpack_userparams( 68 req_param_names=req_params, opt_param_names=opt_param) 69 70 if "AccessPoint" in self.user_params: 71 self.legacy_configure_ap_and_start(ap_count=2, same_ssid=True) 72 73 wutils.wifi_toggle_state(self.dut, True) 74 if "ifs_params" in self.user_params: 75 self.attn_start_db = self.ifs_params[0]["start_db"] 76 self.gateway = self.ifs_params[0]["gateway"] 77 self.roaming_cycle_seconds = self.ifs_params[0][ 78 "roaming_cycle_seconds"] 79 self.total_test_hour = self.ifs_params[0]["total_test_hour"] 80 self.log_capture_period_hour = self.ifs_params[0][ 81 "log_capture_period_hour"] 82 self.on_active_port = self.ifs_params[0]["on_active_port"] 83 asserts.assert_true( 84 len(self.on_active_port) == 2, "Need setup 2 port.") 85 86 self.tcpdump_pid = None 87 utils.set_location_service(self.dut, True) 88 89 def setup_test(self): 90 super().setup_test() 91 self.dut.droid.wakeLockAcquireBright() 92 self.dut.droid.wakeUpNow() 93 self.dut.unlock_screen() 94 self.tcpdump_pid = start_tcpdump(self.dut, self.test_name) 95 96 def teardown_class(self): 97 self.dut.ed.clear_all_events() 98 99 def teardown_test(self): 100 super().teardown_test() 101 self.dut.droid.wakeLockRelease() 102 self.dut.droid.goToSleepNow() 103 wutils.reset_wifi(self.dut) 104 105 def simulate_roaming(self): 106 """ 107 To simulate user move between ap1 and ap2: 108 109 1. Move to ap2: 110 Set ap1's signal attenuation gradually changed from 0 to max_db 111 Set ap2's signal attenuation gradually changed from start_db to 0 112 113 2. Then move to ap1: 114 Set ap1's signal attenuation gradually changed from start_db to 0 115 Set ap2's signal attenuation gradually changed from 0 to max_db 116 117 * 0<start_db<max_db 118 """ 119 attn_max = 95 120 attn_min = 0 121 122 #on_active_port value should between [0-1,2-3] 123 active_attenuator = { 124 "1": self.attenuators[self.on_active_port[0]], 125 "2": self.attenuators[self.on_active_port[1]] 126 } 127 128 for attenuator in self.attenuators: 129 attenuator.set_atten(attn_max) 130 131 self.simulation_thread_running = True 132 while self.simulation_thread_running: 133 active_attenuator["1"].set_atten(attn_min) 134 active_attenuator["2"].set_atten(attn_max) 135 self.log_attens() 136 time.sleep(10) 137 138 active_attenuator["2"].set_atten(self.start_db) 139 self.log_attens() 140 time.sleep(5) 141 for i in range(self.roaming_cycle_seconds): 142 db1 = math.ceil(attn_max / self.roaming_cycle_seconds * 143 (i + 1)) 144 db2 = self.start_db - math.ceil( 145 self.start_db / self.roaming_cycle_seconds * (i + 1)) 146 active_attenuator["1"].set_atten(db1) 147 active_attenuator["2"].set_atten(db2) 148 self.log_attens() 149 time.sleep(1) 150 151 active_attenuator["1"].set_atten(self.start_db) 152 self.log_attens() 153 time.sleep(5) 154 for i in range(self.roaming_cycle_seconds): 155 db1 = math.ceil(attn_max / self.roaming_cycle_seconds * 156 (i + 1)) 157 db2 = self.start_db - math.ceil( 158 self.start_db / self.roaming_cycle_seconds * (i + 1)) 159 active_attenuator["1"].set_atten(db2) 160 active_attenuator["2"].set_atten(db1) 161 self.log_attens() 162 time.sleep(1) 163 self.atten_roaming_count += 1 164 165 def catch_log(self): 166 """Capture logs include bugreport, ANR, mount,ps,vendor,tcpdump""" 167 168 self.log.info("Get log for regular capture.") 169 file_name = time.strftime("%Y-%m-%d_%H:%M:%S", time.localtime()) 170 current_path = os.path.join(self.dut.log_path, file_name) 171 os.makedirs(current_path, exist_ok=True) 172 serial_number = self.dut.serial 173 174 try: 175 out = self.dut.adb.shell("bugreportz", timeout=240) 176 if not out.startswith("OK"): 177 raise AndroidDeviceError( 178 'Failed to take bugreport on %s: %s' % (serial_number, 179 out), 180 serial=serial_number) 181 br_out_path = out.split(':')[1].strip().split()[0] 182 self.dut.adb.pull("%s %s" % (br_out_path, self.dut.log_path)) 183 self.dut.adb.pull("/data/anr {}".format(current_path), timeout=600) 184 self.dut.adb._exec_adb_cmd("shell", "mount > {}".format( 185 os.path.join(current_path, "mount.txt"))) 186 self.dut.adb._exec_adb_cmd("shell", "ps > {}".format( 187 os.path.join(current_path, "ps.txt"))) 188 self.dut.adb.pull("/data/misc/logd {}".format(current_path)) 189 self.dut.adb.pull( 190 "/data/vendor {}".format(current_path), timeout=800) 191 stop_tcpdump( 192 self.dut, self.tcpdump_pid, file_name, adb_pull_timeout=600) 193 self.tcpdump_pid = start_tcpdump(self.dut, file_name) 194 except TimeoutError as e: 195 self.log.error(e) 196 197 def http_request(self, url="https://www.google.com/"): 198 """Get the result via string from target url 199 200 Args: 201 url: target url to loading 202 203 Returns: 204 True if http_request pass 205 """ 206 207 self.ping_count += 1 208 try: 209 self.dut.droid.httpRequestString(url) 210 self.log.info("httpRequest Finish") 211 time.sleep(1) 212 return True 213 except rpc_client.Sl4aApiError as e: 214 self.log.warning("httpRequest Fail.") 215 self.log.warning(e) 216 # Set check delay if http request fail during device roaming. 217 # Except finish roaming within 10s. 218 time.sleep(10) 219 self.log.warning("Ping Google DNS response : {}".format( 220 self.can_ping("8.8.8.8"))) 221 for gate in self.gateway: 222 ping_result = self.can_ping(gate) 223 self.log.warning("Ping AP Gateway[{}] response : {}".format( 224 gate, ping_result)) 225 if ping_result: 226 self.retry_pass_count += 1 227 return True 228 self.fail_count += 1 229 return False 230 231 def log_attens(self): 232 """Log DB from channels""" 233 234 attenuation = ', '.join('{:>5.2f}dB '.format(atten.get_atten()) 235 for atten in self.attenuators) 236 self.log.debug('[Attenuation] %s', attenuation) 237 238 def can_ping(self, ip_addr): 239 """A function to check ping pass. 240 241 Args: 242 ip_addr: target ip address to ping 243 244 Returns: 245 True if ping pass 246 """ 247 ping_result = self.dut.adb.shell("ping -c 1 {}".format(ip_addr)) 248 return '0%' in ping_result.split(' ') 249 250 def browsing_test(self, stress_hour_time): 251 """Continue stress http_request and capture log if any fail 252 253 Args: 254 stress_hour_time: hour of time to stress http_request 255 """ 256 t = threading.Thread(target=self.simulate_roaming) 257 t.start() 258 start_time = time.time() 259 http_request_failed = False 260 while time.time() < start_time + stress_hour_time * 3600: 261 if not self.http_request(): 262 http_request_failed = True 263 self.simulation_thread_running = False 264 t.join() 265 if http_request_failed: 266 self.catch_log() 267 else: 268 stop_standing_subprocess(self.tcpdump_pid) 269 file_name = time.strftime("%Y-%m-%d_%H:%M:%S", time.localtime()) 270 self.tcpdump_pid = start_tcpdump(self.dut, file_name) 271 272 def test_roaming(self): 273 network = self.reference_networks[0]["2g"] 274 wutils.connect_to_wifi_network(self.dut, network) 275 276 time.sleep(10) 277 test_time_slot = int( 278 self.total_test_hour / self.log_capture_period_hour) 279 edge_time_slot = int( 280 self.total_test_hour % self.log_capture_period_hour) 281 282 for i in range(test_time_slot): 283 self.browsing_test(self.log_capture_period_hour) 284 if edge_time_slot: 285 self.browsing_test(edge_time_slot) 286 287 self.log.info("Total roaming times: {}".format( 288 self.atten_roaming_count)) 289 self.log.info("Total ping times: {}".format(self.ping_count)) 290 self.log.info("Retry pass times: {}".format(self.retry_pass_count)) 291 self.log.info("Total fail times: {}".format(self.fail_count)) 292 if self.fail_count: 293 signals.TestFailure( 294 'Find roaming fail condition', 295 extras={ 296 'Roaming fail times': self.fail_count 297 }) 298