1#!/usr/bin/env python3.4
2#
3#   Copyright 2016 - 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
17import itertools
18import queue
19
20from acts import asserts
21from acts import base_test
22from acts import utils
23from acts.test_decorators import test_tracker_info
24from acts.test_utils.wifi import wifi_test_utils as wutils
25
26BSSID_EVENT_WAIT = 30
27
28BSSID_EVENT_TAG = "WifiScannerBssid"
29SCAN_EVENT_TAG = "WifiScannerScan"
30SCANTIME = 10000  #framework support only 10s as minimum scan interval
31
32
33class WifiScannerBssidError(Exception):
34    pass
35
36
37class WifiScannerBssidTest(base_test.BaseTestClass):
38    def __init__(self, controllers):
39        base_test.BaseTestClass.__init__(self, controllers)
40        # A list of all test cases to be executed in this class.
41        self.tests = ("test_wifi_track_bssid_sanity",
42                      "test_wifi_track_bssid_found",
43                      "test_wifi_track_bssid_lost",
44                      "test_wifi_track_bssid_for_2g_while_scanning_5g_channels",
45                      "test_wifi_track_bssid_for_5g_while_scanning_2g_channels",)
46
47    def setup_class(self):
48        self.default_scan_setting = {
49            "band": wutils.WifiEnums.WIFI_BAND_BOTH_WITH_DFS,
50            "periodInMs": SCANTIME,
51            "reportEvents": wutils.WifiEnums.REPORT_EVENT_AFTER_EACH_SCAN,
52            'numBssidsPerScan': 32
53        }
54        self.leeway = 5
55        self.stime_channel = 47  #dwell time plus 2ms
56        self.dut = self.android_devices[0]
57        wutils.wifi_test_device_init(self.dut)
58        self.attenuators = wutils.group_attenuators(self.attenuators)
59        asserts.assert_true(self.dut.droid.wifiIsScannerSupported(),
60                            "Device %s doesn't support WifiScanner, abort." %
61                            self.dut.model)
62        """It will setup the required dependencies and fetch the user params from
63          config file"""
64        self.attenuators[0].set_atten(0)
65        self.attenuators[1].set_atten(0)
66        req_params = ("bssid_2g", "bssid_5g", "bssid_dfs", "attenuator_id",
67                      "max_bugreports")
68        self.wifi_chs = wutils.WifiChannelUS(self.dut.model)
69        self.unpack_userparams(req_params, two_ap_testbed=False)
70
71    def teardown_class(self):
72        BaseTestClass.teardown_test(self)
73        self.log.debug("Shut down all wifi scanner activities.")
74        self.dut.droid.wifiScannerShutdown()
75
76    def on_fail(self, test_name, begin_time):
77        if self.max_bugreports > 0:
78            self.dut.take_bug_report(test_name, begin_time)
79            self.max_bugreports -= 1
80
81    """ Helper Functions Begin """
82
83    def fetch_scan_result(self, scan_idx, scan_setting):
84        """Fetch the scan result for provider listener index.
85
86        This function calculate the time required for scanning based on scan setting
87        and wait for scan result event, on triggering of event process the scan result.
88
89        Args:
90          scan_idx: Index of the scan listener.
91          scan_setting: Setting used for starting the scan.
92
93        Returns:
94          scan_results: if scan result available.
95        """
96        #generating event wait time from scan setting plus leeway
97        self.log.debug(scan_setting)
98        scan_time, scan_channels = wutils.get_scan_time_and_channels(
99            self.wifi_chs, scan_setting, self.stime_channel)
100        scan_time += scan_setting['periodInMs'
101                                  ]  #add scan period delay for next cycle
102        if scan_setting[
103                "reportEvents"] == wutils.WifiEnums.REPORT_EVENT_AFTER_EACH_SCAN:
104            waittime = int(scan_time / 1000) + self.leeway
105        else:
106            time_cache = scan_setting['periodInMs'] * 10  #default cache
107            waittime = int((time_cache + scan_time) / 1000) + self.leeway
108        event_name = "%s%sonResults" % (SCAN_EVENT_TAG, scan_idx)
109        self.log.info("Waiting for the scan result event %s", event_name)
110        event = self.dut.ed.pop_event(event_name, waittime)
111        results = event["data"]["Results"]
112        if len(results) > 0 and "ScanResults" in results[0]:
113            return results[0]["ScanResults"]
114
115    def start_scan_and_validate_environment(self, scan_setting,
116                                            bssid_settings):
117        """Validate environment for test using current scan result for provided
118           settings.
119
120        This function start the scan for given setting and verify that interested
121        Bssids are in scan result or not.
122
123        Args:
124            scan_setting: Setting used for starting the scan.
125            bssid_settings: list of bssid settings.
126
127        Returns:
128            True, if bssid not found in scan result.
129        """
130        try:
131            data = wutils.start_wifi_background_scan(self.dut, scan_setting)
132            self.scan_idx = data["Index"]
133            results = self.fetch_scan_result(self.scan_idx, scan_setting)
134            self.log.debug("scan result %s.", results)
135            asserts.assert_true(results,
136                                "Device is not able to fetch the scan results")
137            for result in results:
138                for bssid_setting in bssid_settings:
139                    if bssid_setting[wutils.WifiEnums.BSSID_KEY] == result[
140                            wutils.WifiEnums.BSSID_KEY]:
141                        asserts.fail(("Test environment is not valid: Bssid %s"
142                                      "already exist in current scan results")
143                                     % result[wutils.WifiEnums.BSSID_KEY])
144        except queue.Empty as error:
145            self.dut.droid.wifiScannerStopBackgroundScan(self.scan_idx)
146            raise AssertionError(
147                "OnResult event did not triggered for scanner\n%s" % error)
148
149    def check_bssid_in_found_result(self, bssid_settings, found_results):
150        """look for any tracked bssid in reported result of found bssids.
151
152        Args:
153            bssid_settings:Setting used for tracking bssids.
154            found_results: Result reported in found event.
155
156        Returns:
157            True if bssid is present in result.
158        """
159        for bssid_setting in bssid_settings:
160            for found_result in found_results:
161                if found_result[wutils.WifiEnums.BSSID_KEY] == bssid_setting[
162                        wutils.WifiEnums.BSSID_KEY]:
163                    return
164        asserts.fail("Test fail because Bssid %s is not found in event results"
165                     % bssid_settings)
166
167    def track_bssid_with_vaild_scan_for_found(self, track_setting):
168        """Common logic for tracking a bssid for Found event.
169
170         1. Starts Wifi Scanner bssid tracking for interested bssids in track_setting.
171         2. Start Wifi Scanner scan with default scan settings.
172         3. Validate the environment to check AP is not in range.
173         4. Attenuate the signal to make AP in range.
174         5. Verified that onFound event is triggered for interested bssids in
175            track setting.
176
177        Args:
178            track_setting: Setting for bssid tracking.
179
180        Returns:
181            True if found event occur for interested BSSID.
182        """
183        self.attenuators[self.attenuator_id].set_atten(90)
184        data = wutils.start_wifi_track_bssid(self.dut, track_setting)
185        idx = data["Index"]
186        self.start_scan_and_validate_environment(self.default_scan_setting,
187                                                 track_setting["bssidInfos"])
188        try:
189            self.attenuators[self.attenuator_id].set_atten(0)
190            event_name = "%s%sonFound" % (BSSID_EVENT_TAG, idx)
191            self.log.info("Waiting for the BSSID event %s", event_name)
192            event = self.dut.ed.pop_event(event_name, BSSID_EVENT_WAIT)
193            self.log.debug(event)
194            self.check_bssid_in_found_result(track_setting["bssidInfos"],
195                                             event["data"]["Results"])
196        except queue.Empty as error:
197            self.log.error(error)
198            # log scan result for debugging
199            results = self.fetch_scan_result(self.scan_idx,
200                                             self.default_scan_setting)
201            self.log.debug("scan result %s", results)
202            raise AssertionError("Event %s did not triggered for %s\n%s" %
203                                 (event_name, track_setting["bssidInfos"],
204                                  error))
205        finally:
206            self.dut.droid.wifiScannerStopBackgroundScan(self.scan_idx)
207            self.dut.droid.wifiScannerStopTrackingBssids(idx)
208
209    def track_bssid_with_vaild_scan_for_lost(self, track_setting):
210        """Common logic for tracking a bssid for Lost event.
211
212         1. Start Wifi Scanner scan with default scan settings.
213         2. Validate the environment to check AP is not in range.
214         3. Starts Wifi Scanner bssid tracking for interested bssids in track_setting.
215         4. Attenuate the signal to make Bssids in range.
216         5. Verified that onFound event is triggered for interested bssids in
217            track setting.
218         6. Attenuate the signal to make Bssids out of range.
219         7. Verified that onLost event is triggered.
220
221        Args:
222            track_setting: Setting for bssid tracking.
223            scan_setting: Setting used for starting the scan.
224
225        Returns:
226            True if Lost event occur for interested BSSID.
227        """
228        self.attenuators[self.attenuator_id].set_atten(90)
229        self.start_scan_and_validate_environment(self.default_scan_setting,
230                                                 track_setting["bssidInfos"])
231        idx = None
232        found = False
233        try:
234            data = wutils.start_wifi_track_bssid(self.dut, track_setting)
235            idx = data["Index"]
236            self.attenuators[self.attenuator_id].set_atten(0)
237            #onFound event should be occurre before tracking for onLost event
238            event_name = "%s%sonFound" % (BSSID_EVENT_TAG, idx)
239            self.log.info("Waiting for the BSSID event %s", event_name)
240            event = self.dut.ed.pop_event(event_name, BSSID_EVENT_WAIT)
241            self.log.debug(event)
242            self.check_bssid_in_found_result(track_setting["bssidInfos"],
243                                             event["data"]["Results"])
244            self.attenuators[self.attenuator_id].set_atten(90)
245            # log scan result for debugging
246            for i in range(1, track_setting["apLostThreshold"]):
247                results = self.fetch_scan_result(self.scan_idx,
248                                                 self.default_scan_setting)
249                self.log.debug("scan result %s %s", i, results)
250            event_name = "%s%sonLost" % (BSSID_EVENT_TAG, idx)
251            self.log.info("Waiting for the BSSID event %s", event_name)
252            event = self.dut.ed.pop_event(event_name, BSSID_EVENT_WAIT)
253            self.log.debug(event)
254        except queue.Empty as error:
255            raise AssertionError("Event %s did not triggered for %s\n%s" %
256                                 (event_name, track_setting["bssidInfos"],
257                                  error))
258        finally:
259            self.dut.droid.wifiScannerStopBackgroundScan(self.scan_idx)
260            if idx:
261                self.dut.droid.wifiScannerStopTrackingBssids(idx)
262
263    def wifi_generate_track_bssid_settings(self, isLost):
264        """Generates all the combinations of different track setting parameters.
265
266        Returns:
267            A list of dictionaries each representing a set of track settings.
268        """
269        bssids = [[self.bssid_2g], [self.bssid_5g],
270                  [self.bssid_2g, self.bssid_5g]]
271        if self.dut.model != "hammerhead" or not self.two_ap_testbed:
272            bssids.append([self.bssid_dfs])
273        if isLost:
274            apthreshold = (3, 5)
275        else:
276            apthreshold = (1, )
277        # Create track setting strings based on the combinations
278        setting_combinations = list(itertools.product(bssids, apthreshold))
279        # Create scan setting strings based on the combinations
280        track_settings = []
281        for combo in setting_combinations:
282            s = {}
283            s["bssidInfos"] = combo[0]
284            s["apLostThreshold"] = combo[1]
285            track_settings.append(s)
286        return track_settings
287
288    def track_setting_to_string(self, track_setting):
289        """Convert track setting to string for Bssids in that"""
290        string = ""
291        for bssid_setting in track_setting:
292            string += bssid_setting[wutils.WifiEnums.BSSID_KEY]
293            string += "_"
294        return string
295
296    def combineBssids(self, *track_settings):
297        """Combine bssids in the track_settings to one list"""
298        bssids = []
299        for track_setting in track_settings:
300            bssids.extend(track_setting["bssidInfos"])
301        return bssids
302
303    """ Helper Functions End """
304    """ Tests Begin """
305
306    @test_tracker_info(uuid="599a30b8-73ad-4314-a245-7ec58fc7e74b")
307    def test_wifi_track_bssid_found(self):
308        """Test bssid track for event found with a list of different settings.
309
310         1. Starts Wifi Scanner bssid tracking for interested bssids in track_setting.
311         2. Start Wifi Scanner scan with default scan settings.
312         3. Validate the environment to check AP is not in range.
313         4. Attenuate the signal to make AP in range.
314         5. Verified that onFound event is triggered for interested bssids in
315            track setting.
316        """
317        track_settings = self.wifi_generate_track_bssid_settings(False)
318        name_func = lambda track_setting: "test_wifi_track_found_bssidInfos_%sapLostThreshold_%s" % (self.track_setting_to_string(track_setting["bssidInfos"]), track_setting["apLostThreshold"])
319        failed = self.run_generated_testcases(
320            self.track_bssid_with_vaild_scan_for_found,
321            track_settings,
322            name_func=name_func)
323        asserts.assert_false(
324            failed, "Track bssid found failed with these bssids: %s" % failed)
325
326    @test_tracker_info(uuid="7ebd4b61-c408-45b3-b9b6-098753d46aa7")
327    def test_wifi_track_bssid_lost(self):
328        """Test bssid track for event lost with a list of different settings.
329
330         1. Start Wifi Scanner scan with default scan settings.
331         2. Validate the environment to check AP is not in range.
332         3. Starts Wifi Scanner bssid tracking for interested bssids in track_setting.
333         4. Attenuate the signal to make Bssids in range.
334         5. Verified that onFound event is triggered for interested bssids in
335            track setting.
336         6. Attenuate the signal to make Bssids out of range.
337         7. Verified that onLost event is triggered.
338        """
339        track_settings = self.wifi_generate_track_bssid_settings(True)
340        name_func = lambda track_setting: "test_wifi_track_lost_bssidInfos_%sapLostThreshold_%s" % (self.track_setting_to_string(track_setting["bssidInfos"]), track_setting["apLostThreshold"])
341        failed = self.run_generated_testcases(
342            self.track_bssid_with_vaild_scan_for_lost,
343            track_settings,
344            name_func=name_func)
345        asserts.assert_false(
346            failed, "Track bssid lost failed with these bssids: %s" % failed)
347
348    def test_wifi_track_bssid_sanity(self):
349        """Test bssid track for event found and lost with default settings.
350
351         1. Start WifiScanner scan for default scan settings.
352         2. Start Bssid track for "bssid_2g" AP.
353         3. Attenuate the signal to move in AP range.
354         4. Verify that onFound event occur.
355         5. Attenuate the signal to move out of range
356         6. Verify that onLost event occur.
357        """
358        track_setting = {"bssidInfos": [self.bssid_2g], "apLostThreshold": 3}
359        self.track_bssid_with_vaild_scan_for_lost(track_setting)
360
361    def test_wifi_track_bssid_for_2g_while_scanning_5g_channels(self):
362        """Test bssid track for 2g bssids while scanning 5g channels.
363
364         1. Starts Wifi Scanner bssid tracking for 2g bssids in track_setting.
365         2. Start Wifi Scanner scan for 5G Band only.
366         3. Validate the environment to check AP is not in range.
367         4. Attenuate the signal to make AP in range.
368         5. Verified that onFound event isn't triggered for 2g bssids.
369      """
370        self.attenuators[self.attenuator_id].set_atten(90)
371        scan_setting = {"band": wutils.WifiEnums.WIFI_BAND_5_GHZ,
372                        "periodInMs": SCANTIME,
373                        "reportEvents":
374                        wutils.WifiEnums.REPORT_EVENT_AFTER_EACH_SCAN,
375                        "numBssidsPerScan": 32}
376        track_setting = {"bssidInfos": [self.bssid_2g], "apLostThreshold": 3}
377        self.start_scan_and_validate_environment(scan_setting,
378                                                 track_setting["bssidInfos"])
379        idx = None
380        try:
381            data = wutils.start_wifi_track_bssid(self.dut, track_setting)
382            idx = data["Index"]
383            self.attenuators[self.attenuator_id].set_atten(0)
384            event_name = "%s%sonFound" % (BSSID_EVENT_TAG, idx)
385            self.log.info("Waiting for the BSSID event %s", event_name)
386            #waiting for 2x time to make sure
387            event = self.dut.ed.pop_event(event_name, BSSID_EVENT_WAIT * 2)
388            self.log.debug(event)
389            self.check_bssid_in_found_result(track_setting["bssidInfos"],
390                                             event["data"]["Results"])
391        except queue.Empty as error:
392            self.log.info(
393                "As excepted event didn't occurred with different scan setting")
394        finally:
395            self.dut.droid.wifiScannerStopBackgroundScan(self.scan_idx)
396            if idx:
397                self.dut.droid.wifiScannerStopTrackingBssids(idx)
398
399    def test_wifi_track_bssid_for_5g_while_scanning_2g_channels(self):
400        """Test bssid track for 5g bssids while scanning 2g channels.
401
402           1. Starts Wifi Scanner bssid tracking for 5g bssids in track_setting.
403           2. Start Wifi Scanner scan for 2G Band only.
404           3. Validate the environment to check AP is not in range.
405           4. Attenuate the signal to make AP in range.
406           5. Verified that onFound event isn't triggered for 5g bssids.
407        """
408        self.attenuators[self.attenuator_id].set_atten(90)
409        scan_setting = {"band": wutils.WifiEnums.WIFI_BAND_24_GHZ,
410                        "periodInMs": SCANTIME,
411                        "reportEvents":
412                        wutils.WifiEnums.REPORT_EVENT_AFTER_EACH_SCAN,
413                        "numBssidsPerScan": 32}
414        track_setting = {"bssidInfos": [self.bssid_5g], "apLostThreshold": 3}
415        data = wutils.start_wifi_track_bssid(self.dut, track_setting)
416        idx = data["Index"]
417        self.start_scan_and_validate_environment(scan_setting,
418                                                 track_setting["bssidInfos"])
419        try:
420            self.attenuators[self.attenuator_id].set_atten(0)
421            event_name = "%s%sonFound" % (BSSID_EVENT_TAG, idx)
422            self.log.info("Waiting for the BSSID event %s", event_name)
423            #waiting for 2x time to make sure
424            event = self.dut.ed.pop_event(event_name, BSSID_EVENT_WAIT * 2)
425            self.log.debug(event)
426            self.check_bssid_in_found_result(track_setting["bssidInfos"],
427                                             event["data"]["Results"])
428        except queue.Empty as error:
429            self.log.info(
430                "As excepted event didn't occurred with different scan setting")
431        finally:
432            self.dut.droid.wifiScannerStopBackgroundScan(self.scan_idx)
433            if idx:
434                self.dut.droid.wifiScannerStopTrackingBssids(idx)
435
436    def test_wifi_tracking_bssid_multi_listeners_found(self):
437        """Test bssid tracking for multiple listeners
438            1. Start BSSID tracking for 5g bssids
439            2. Start BSSID tracking for 2g bssids
440            3. Start WifiScanner scan on both bands.
441            4. Valid the environment and check the APs are not in range.
442            5. Attenuate the signal to make the APs in range.
443            6. Verify onFound event triggered on both APs.
444        """
445        # Attenuate the signal to make APs invisible.
446        self.attenuators[self.attenuator_id].set_atten(90)
447        scan_setting = { "band": WifiEnums.WIFI_BAND_BOTH_WITH_DFS,
448                         "periodInMs": SCANTIME,
449                         "reportEvents": WifiEnums.REPORT_EVENT_AFTER_EACH_SCAN,
450                         "numBssidsPerScan": 32}
451        track_setting_5g = {"bssidInfos":[self.bssid_5g], "apLostThreshold":3}
452        data_5g = start_wifi_track_bssid(self.dut, track_setting_5g)
453        idx_5g = data_5g["Index"]
454
455        track_setting_2g = {"bssidInfos":[self.bssid_2g], "apLostThreshold":3}
456        data_2g = start_wifi_track_bssid(self.dut, track_setting_2g)
457        idx_2g = data_2g["Index"]
458
459        valid_env = self.start_scan_and_validate_environment(
460            scan_setting, self.combineBssids(track_setting_5g, track_setting_2g))
461        try:
462            asserts.assert_true(valid_env,
463                                "Test environment is not valid, AP is in range")
464            self.attenuators[self.attenuator_id].set_atten(0)
465            event_name = "{}{}{}{}onFound".format(BSSID_EVENT_TAG, idx_5g, BSSID_EVENT_TAG, idx_2g)
466            self.log.info("Waiting for the BSSID event {}".format(event_name))
467            #waiting for 2x time to make sure
468            event = self.dut.ed.pop_event(event_name, BSSID_EVENT_WAIT * 2)
469            self.log.debug(event)
470            found = self.check_bssid_in_found_result(
471                self.combineBssids(track_setting_5g, track_setting_2g),
472                event["data"]["Results"])
473            asserts.assert_true(found,
474                                "Test failed because Bssid onFound event is not triggered")
475        finally:
476            self.dut.droid.wifiScannerStopBackgroundScan(self.scan_idx)
477            if idx_5g:
478                self.dut.droid.wifiScannerStopTrackingBssids(idx_5g)
479            if idx_2g:
480                self.dut.droid.wifiScannerStopTrackingBssids(idx_2g);
481
482""" Tests End """
483