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 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        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)
169        os.makedirs(current_path, exist_ok=True)
170        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                })
296