blob: 5b1603dd4ad0ae0455406c7da572e1ce090c7eb7 [file] [log] [blame]
jerrypcchen@google.com91dab8c2018-11-21 15:49:57 +08001#
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
Xianyuan Jia168103b2019-09-06 12:22:52 -070046 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()
jerrypcchen@google.com91dab8c2018-11-21 15:49:57 +080054 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
jerrypcchen@google.com91dab8c2018-11-21 15:49:57 +080062 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 self.dut.droid.wakeLockAcquireBright()
91 self.dut.droid.wakeUpNow()
92 self.dut.unlock_screen()
93 self.tcpdump_pid = start_tcpdump(self.dut, self.test_name)
94
95 def teardown_class(self):
96 self.dut.ed.clear_all_events()
97
98 def teardown_test(self):
99 self.dut.droid.wakeLockRelease()
100 self.dut.droid.goToSleepNow()
101 wutils.reset_wifi(self.dut)
102
103 def simulate_roaming(self):
104 """
105 To simulate user move between ap1 and ap2:
106
107 1. Move to ap2:
108 Set ap1's signal attenuation gradually changed from 0 to max_db
109 Set ap2's signal attenuation gradually changed from start_db to 0
110
111 2. Then move to ap1:
112 Set ap1's signal attenuation gradually changed from start_db to 0
113 Set ap2's signal attenuation gradually changed from 0 to max_db
114
115 * 0<start_db<max_db
116 """
117 attn_max = 95
118 attn_min = 0
119
120 #on_active_port value should between [0-1,2-3]
121 active_attenuator = {
122 "1": self.attenuators[self.on_active_port[0]],
123 "2": self.attenuators[self.on_active_port[1]]
124 }
125
126 for attenuator in self.attenuators:
127 attenuator.set_atten(attn_max)
128
129 self.simulation_thread_running = True
130 while self.simulation_thread_running:
131 active_attenuator["1"].set_atten(attn_min)
132 active_attenuator["2"].set_atten(attn_max)
133 self.log_attens()
134 time.sleep(10)
135
136 active_attenuator["2"].set_atten(self.start_db)
137 self.log_attens()
138 time.sleep(5)
139 for i in range(self.roaming_cycle_seconds):
140 db1 = math.ceil(attn_max / self.roaming_cycle_seconds *
141 (i + 1))
142 db2 = self.start_db - math.ceil(
143 self.start_db / self.roaming_cycle_seconds * (i + 1))
144 active_attenuator["1"].set_atten(db1)
145 active_attenuator["2"].set_atten(db2)
146 self.log_attens()
147 time.sleep(1)
148
149 active_attenuator["1"].set_atten(self.start_db)
150 self.log_attens()
151 time.sleep(5)
152 for i in range(self.roaming_cycle_seconds):
153 db1 = math.ceil(attn_max / self.roaming_cycle_seconds *
154 (i + 1))
155 db2 = self.start_db - math.ceil(
156 self.start_db / self.roaming_cycle_seconds * (i + 1))
157 active_attenuator["1"].set_atten(db2)
158 active_attenuator["2"].set_atten(db1)
159 self.log_attens()
160 time.sleep(1)
161 self.atten_roaming_count += 1
162
163 def catch_log(self):
164 """Capture logs include bugreport, ANR, mount,ps,vendor,tcpdump"""
165
166 self.log.info("Get log for regular capture.")
167 file_name = time.strftime("%Y-%m-%d_%H:%M:%S", time.localtime())
168 current_path = os.path.join(self.dut.log_path, file_name)
Mark De Ruyter72f8df92020-02-12 13:44:49 -0800169 os.makedirs(current_path, exist_ok=True)
jerrypcchen@google.com91dab8c2018-11-21 15:49:57 +0800170 serial_number = self.dut.serial
171
172 try:
173 out = self.dut.adb.shell("bugreportz", timeout=240)
174 if not out.startswith("OK"):
175 raise AndroidDeviceError(
176 'Failed to take bugreport on %s: %s' % (serial_number,
177 out),
178 serial=serial_number)
179 br_out_path = out.split(':')[1].strip().split()[0]
180 self.dut.adb.pull("%s %s" % (br_out_path, self.dut.log_path))
181 self.dut.adb.pull("/data/anr {}".format(current_path), timeout=600)
182 self.dut.adb._exec_adb_cmd("shell", "mount > {}".format(
183 os.path.join(current_path, "mount.txt")))
184 self.dut.adb._exec_adb_cmd("shell", "ps > {}".format(
185 os.path.join(current_path, "ps.txt")))
186 self.dut.adb.pull("/data/misc/logd {}".format(current_path))
187 self.dut.adb.pull(
188 "/data/vendor {}".format(current_path), timeout=800)
189 stop_tcpdump(
190 self.dut, self.tcpdump_pid, file_name, adb_pull_timeout=600)
191 self.tcpdump_pid = start_tcpdump(self.dut, file_name)
192 except TimeoutError as e:
193 self.log.error(e)
194
195 def http_request(self, url="https://www.google.com/"):
196 """Get the result via string from target url
197
198 Args:
199 url: target url to loading
200
201 Returns:
202 True if http_request pass
203 """
204
205 self.ping_count += 1
206 try:
207 self.dut.droid.httpRequestString(url)
208 self.log.info("httpRequest Finish")
209 time.sleep(1)
210 return True
211 except rpc_client.Sl4aApiError as e:
212 self.log.warning("httpRequest Fail.")
213 self.log.warning(e)
214 # Set check delay if http request fail during device roaming.
215 # Except finish roaming within 10s.
216 time.sleep(10)
217 self.log.warning("Ping Google DNS response : {}".format(
218 self.can_ping("8.8.8.8")))
219 for gate in self.gateway:
220 ping_result = self.can_ping(gate)
221 self.log.warning("Ping AP Gateway[{}] response : {}".format(
222 gate, ping_result))
223 if ping_result:
224 self.retry_pass_count += 1
225 return True
226 self.fail_count += 1
227 return False
228
229 def log_attens(self):
230 """Log DB from channels"""
231
232 attenuation = ', '.join('{:>5.2f}dB '.format(atten.get_atten())
233 for atten in self.attenuators)
234 self.log.debug('[Attenuation] %s', attenuation)
235
236 def can_ping(self, ip_addr):
237 """A function to check ping pass.
238
239 Args:
240 ip_addr: target ip address to ping
241
242 Returns:
243 True if ping pass
244 """
245 ping_result = self.dut.adb.shell("ping -c 1 {}".format(ip_addr))
246 return '0%' in ping_result.split(' ')
247
248 def browsing_test(self, stress_hour_time):
249 """Continue stress http_request and capture log if any fail
250
251 Args:
252 stress_hour_time: hour of time to stress http_request
253 """
254 t = threading.Thread(target=self.simulate_roaming)
255 t.start()
256 start_time = time.time()
257 http_request_failed = False
258 while time.time() < start_time + stress_hour_time * 3600:
259 if not self.http_request():
260 http_request_failed = True
261 self.simulation_thread_running = False
262 t.join()
263 if http_request_failed:
264 self.catch_log()
265 else:
266 stop_standing_subprocess(self.tcpdump_pid)
267 file_name = time.strftime("%Y-%m-%d_%H:%M:%S", time.localtime())
268 self.tcpdump_pid = start_tcpdump(self.dut, file_name)
269
270 def test_roaming(self):
271 network = self.reference_networks[0]["2g"]
272 wutils.connect_to_wifi_network(self.dut, network)
273
274 time.sleep(10)
275 test_time_slot = int(
276 self.total_test_hour / self.log_capture_period_hour)
277 edge_time_slot = int(
278 self.total_test_hour % self.log_capture_period_hour)
279
280 for i in range(test_time_slot):
281 self.browsing_test(self.log_capture_period_hour)
282 if edge_time_slot:
283 self.browsing_test(edge_time_slot)
284
285 self.log.info("Total roaming times: {}".format(
286 self.atten_roaming_count))
287 self.log.info("Total ping times: {}".format(self.ping_count))
288 self.log.info("Retry pass times: {}".format(self.retry_pass_count))
289 self.log.info("Total fail times: {}".format(self.fail_count))
290 if self.fail_count:
291 signals.TestFailure(
292 'Find roaming fail condition',
293 extras={
294 'Roaming fail times': self.fail_count
295 })