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.test_utils.net.net_test_utils import start_tcpdump, stop_tcpdump 29from acts.test_utils.wifi import wifi_test_utils as wutils 30from acts.test_utils.wifi.wifi_test_utils import WifiEnums 31from acts.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 __init__(self, controllers): 47 WifiBaseTest.__init__(self, controllers) 48 self.simulation_thread_running = False 49 self.atten_roaming_count = 0 50 self.start_db = 30 51 self.roaming_cycle_seconds = 20 52 self.fail_count = 0 53 self.retry_pass_count = 0 54 self.ping_count = 0 55 56 def setup_class(self): 57 """Setup required dependencies from config file and configure 58 the required networks for testing roaming. 59 60 Returns: 61 True if successfully configured the requirements for testing. 62 """ 63 self.dut = self.android_devices[0] 64 wutils.wifi_test_device_init(self.dut) 65 req_params = ["attenuators", "ifs_params"] 66 opt_param = [] 67 68 self.unpack_userparams( 69 req_param_names=req_params, opt_param_names=opt_param) 70 71 if "AccessPoint" in self.user_params: 72 self.legacy_configure_ap_and_start(ap_count=2, same_ssid=True) 73 74 wutils.wifi_toggle_state(self.dut, True) 75 if "ifs_params" in self.user_params: 76 self.attn_start_db = self.ifs_params[0]["start_db"] 77 self.gateway = self.ifs_params[0]["gateway"] 78 self.roaming_cycle_seconds = self.ifs_params[0][ 79 "roaming_cycle_seconds"] 80 self.total_test_hour = self.ifs_params[0]["total_test_hour"] 81 self.log_capture_period_hour = self.ifs_params[0][ 82 "log_capture_period_hour"] 83 self.on_active_port = self.ifs_params[0]["on_active_port"] 84 asserts.assert_true( 85 len(self.on_active_port) == 2, "Need setup 2 port.") 86 87 self.tcpdump_pid = None 88 utils.set_location_service(self.dut, True) 89 90 def setup_test(self): 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 self.dut.droid.wakeLockRelease() 101 self.dut.droid.goToSleepNow() 102 wutils.reset_wifi(self.dut) 103 104 def simulate_roaming(self): 105 """ 106 To simulate user move between ap1 and ap2: 107 108 1. Move to ap2: 109 Set ap1's signal attenuation gradually changed from 0 to max_db 110 Set ap2's signal attenuation gradually changed from start_db to 0 111 112 2. Then move to ap1: 113 Set ap1's signal attenuation gradually changed from start_db to 0 114 Set ap2's signal attenuation gradually changed from 0 to max_db 115 116 * 0<start_db<max_db 117 """ 118 attn_max = 95 119 attn_min = 0 120 121 #on_active_port value should between [0-1,2-3] 122 active_attenuator = { 123 "1": self.attenuators[self.on_active_port[0]], 124 "2": self.attenuators[self.on_active_port[1]] 125 } 126 127 for attenuator in self.attenuators: 128 attenuator.set_atten(attn_max) 129 130 self.simulation_thread_running = True 131 while self.simulation_thread_running: 132 active_attenuator["1"].set_atten(attn_min) 133 active_attenuator["2"].set_atten(attn_max) 134 self.log_attens() 135 time.sleep(10) 136 137 active_attenuator["2"].set_atten(self.start_db) 138 self.log_attens() 139 time.sleep(5) 140 for i in range(self.roaming_cycle_seconds): 141 db1 = math.ceil(attn_max / self.roaming_cycle_seconds * 142 (i + 1)) 143 db2 = self.start_db - math.ceil( 144 self.start_db / self.roaming_cycle_seconds * (i + 1)) 145 active_attenuator["1"].set_atten(db1) 146 active_attenuator["2"].set_atten(db2) 147 self.log_attens() 148 time.sleep(1) 149 150 active_attenuator["1"].set_atten(self.start_db) 151 self.log_attens() 152 time.sleep(5) 153 for i in range(self.roaming_cycle_seconds): 154 db1 = math.ceil(attn_max / self.roaming_cycle_seconds * 155 (i + 1)) 156 db2 = self.start_db - math.ceil( 157 self.start_db / self.roaming_cycle_seconds * (i + 1)) 158 active_attenuator["1"].set_atten(db2) 159 active_attenuator["2"].set_atten(db1) 160 self.log_attens() 161 time.sleep(1) 162 self.atten_roaming_count += 1 163 164 def catch_log(self): 165 """Capture logs include bugreport, ANR, mount,ps,vendor,tcpdump""" 166 167 self.log.info("Get log for regular capture.") 168 file_name = time.strftime("%Y-%m-%d_%H:%M:%S", time.localtime()) 169 current_path = os.path.join(self.dut.log_path, file_name) 170 utils.create_dir(current_path) 171 serial_number = self.dut.serial 172 173 try: 174 out = self.dut.adb.shell("bugreportz", timeout=240) 175 if not out.startswith("OK"): 176 raise AndroidDeviceError( 177 'Failed to take bugreport on %s: %s' % (serial_number, 178 out), 179 serial=serial_number) 180 br_out_path = out.split(':')[1].strip().split()[0] 181 self.dut.adb.pull("%s %s" % (br_out_path, self.dut.log_path)) 182 self.dut.adb.pull("/data/anr {}".format(current_path), timeout=600) 183 self.dut.adb._exec_adb_cmd("shell", "mount > {}".format( 184 os.path.join(current_path, "mount.txt"))) 185 self.dut.adb._exec_adb_cmd("shell", "ps > {}".format( 186 os.path.join(current_path, "ps.txt"))) 187 self.dut.adb.pull("/data/misc/logd {}".format(current_path)) 188 self.dut.adb.pull( 189 "/data/vendor {}".format(current_path), timeout=800) 190 stop_tcpdump( 191 self.dut, self.tcpdump_pid, file_name, adb_pull_timeout=600) 192 self.tcpdump_pid = start_tcpdump(self.dut, file_name) 193 except TimeoutError as e: 194 self.log.error(e) 195 196 def http_request(self, url="https://www.google.com/"): 197 """Get the result via string from target url 198 199 Args: 200 url: target url to loading 201 202 Returns: 203 True if http_request pass 204 """ 205 206 self.ping_count += 1 207 try: 208 self.dut.droid.httpRequestString(url) 209 self.log.info("httpRequest Finish") 210 time.sleep(1) 211 return True 212 except rpc_client.Sl4aApiError as e: 213 self.log.warning("httpRequest Fail.") 214 self.log.warning(e) 215 # Set check delay if http request fail during device roaming. 216 # Except finish roaming within 10s. 217 time.sleep(10) 218 self.log.warning("Ping Google DNS response : {}".format( 219 self.can_ping("8.8.8.8"))) 220 for gate in self.gateway: 221 ping_result = self.can_ping(gate) 222 self.log.warning("Ping AP Gateway[{}] response : {}".format( 223 gate, ping_result)) 224 if ping_result: 225 self.retry_pass_count += 1 226 return True 227 self.fail_count += 1 228 return False 229 230 def log_attens(self): 231 """Log DB from channels""" 232 233 attenuation = ', '.join('{:>5.2f}dB '.format(atten.get_atten()) 234 for atten in self.attenuators) 235 self.log.debug('[Attenuation] %s', attenuation) 236 237 def can_ping(self, ip_addr): 238 """A function to check ping pass. 239 240 Args: 241 ip_addr: target ip address to ping 242 243 Returns: 244 True if ping pass 245 """ 246 ping_result = self.dut.adb.shell("ping -c 1 {}".format(ip_addr)) 247 return '0%' in ping_result.split(' ') 248 249 def browsing_test(self, stress_hour_time): 250 """Continue stress http_request and capture log if any fail 251 252 Args: 253 stress_hour_time: hour of time to stress http_request 254 """ 255 t = threading.Thread(target=self.simulate_roaming) 256 t.start() 257 start_time = time.time() 258 http_request_failed = False 259 while time.time() < start_time + stress_hour_time * 3600: 260 if not self.http_request(): 261 http_request_failed = True 262 self.simulation_thread_running = False 263 t.join() 264 if http_request_failed: 265 self.catch_log() 266 else: 267 stop_standing_subprocess(self.tcpdump_pid) 268 file_name = time.strftime("%Y-%m-%d_%H:%M:%S", time.localtime()) 269 self.tcpdump_pid = start_tcpdump(self.dut, file_name) 270 271 def test_roaming(self): 272 network = self.reference_networks[0]["2g"] 273 wutils.connect_to_wifi_network(self.dut, network) 274 275 time.sleep(10) 276 test_time_slot = int( 277 self.total_test_hour / self.log_capture_period_hour) 278 edge_time_slot = int( 279 self.total_test_hour % self.log_capture_period_hour) 280 281 for i in range(test_time_slot): 282 self.browsing_test(self.log_capture_period_hour) 283 if edge_time_slot: 284 self.browsing_test(edge_time_slot) 285 286 self.log.info("Total roaming times: {}".format( 287 self.atten_roaming_count)) 288 self.log.info("Total ping times: {}".format(self.ping_count)) 289 self.log.info("Retry pass times: {}".format(self.retry_pass_count)) 290 self.log.info("Total fail times: {}".format(self.fail_count)) 291 if self.fail_count: 292 signals.TestFailure( 293 'Find roaming fail condition', 294 extras={ 295 'Roaming fail times': self.fail_count 296 }) 297