1#!/usr/bin/env python3
2#
3#   Copyright 2017 - Google
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 os
18
19from acts import asserts
20from acts import utils
21from acts.base_test import BaseTestClass
22from acts.keys import Config
23from acts_contrib.test_utils.net import net_test_utils as nutils
24from acts_contrib.test_utils.wifi import wifi_test_utils as wutils
25from acts_contrib.test_utils.wifi.aware import aware_const as aconsts
26from acts_contrib.test_utils.wifi.aware import aware_test_utils as autils
27
28
29class AwareBaseTest(BaseTestClass):
30    # message ID counter to make sure all uses are unique
31    msg_id = 0
32
33    # offset (in seconds) to separate the start-up of multiple devices.
34    # De-synchronizes the start-up time so that they don't start and stop scanning
35    # at the same time - which can lead to very long clustering times.
36    device_startup_offset = 2
37
38    def setup_class(self):
39        opt_param = ["pixel_models", "cnss_diag_file"]
40        self.unpack_userparams(opt_param_names=opt_param)
41        if hasattr(self, "cnss_diag_file"):
42            if isinstance(self.cnss_diag_file, list):
43                self.cnss_diag_file = self.cnss_diag_file[0]
44            if not os.path.isfile(self.cnss_diag_file):
45                self.cnss_diag_file = os.path.join(
46                    self.user_params[Config.key_config_path.value],
47                    self.cnss_diag_file)
48
49    def setup_test(self):
50        required_params = ("aware_default_power_mode",
51                           "dbs_supported_models",)
52        self.unpack_userparams(required_params)
53
54        if hasattr(self, "cnss_diag_file") and hasattr(self, "pixel_models"):
55            wutils.start_cnss_diags(
56                self.android_devices, self.cnss_diag_file, self.pixel_models)
57        self.tcpdump_proc = []
58        if hasattr(self, "android_devices"):
59            for ad in self.android_devices:
60                proc = nutils.start_tcpdump(ad, self.test_name)
61                self.tcpdump_proc.append((ad, proc))
62
63        for ad in self.android_devices:
64            ad.droid.wifiEnableVerboseLogging(1)
65            asserts.skip_if(
66                not ad.droid.doesDeviceSupportWifiAwareFeature(),
67                "Device under test does not support Wi-Fi Aware - skipping test"
68            )
69            aware_avail = ad.droid.wifiIsAwareAvailable()
70            ad.droid.wifiP2pClose()
71            wutils.wifi_toggle_state(ad, True)
72            utils.set_location_service(ad, True)
73            if not aware_avail:
74                self.log.info('Aware not available. Waiting ...')
75                autils.wait_for_event(ad,
76                                      aconsts.BROADCAST_WIFI_AWARE_AVAILABLE)
77            ad.aware_capabilities = autils.get_aware_capabilities(ad)
78            self.reset_device_parameters(ad)
79            self.reset_device_statistics(ad)
80            self.set_power_mode_parameters(ad)
81            wutils.set_wifi_country_code(ad, wutils.WifiEnums.CountryCode.US)
82            autils.configure_ndp_allow_any_override(ad, True)
83            # set randomization interval to 0 (disable) to reduce likelihood of
84            # interference in tests
85            autils.configure_mac_random_interval(ad, 0)
86            ad.ed.clear_all_events()
87
88    def teardown_test(self):
89        if hasattr(self, "cnss_diag_file") and hasattr(self, "pixel_models"):
90            wutils.stop_cnss_diags(self.android_devices, self.pixel_models)
91        for proc in self.tcpdump_proc:
92            nutils.stop_tcpdump(
93                    proc[0], proc[1], self.test_name, pull_dump=False)
94        self.tcpdump_proc = []
95        for ad in self.android_devices:
96            if not ad.droid.doesDeviceSupportWifiAwareFeature():
97                return
98            ad.droid.wifiP2pClose()
99            ad.droid.wifiAwareDestroyAll()
100            self.reset_device_parameters(ad)
101            autils.validate_forbidden_callbacks(ad)
102
103    def reset_device_parameters(self, ad):
104        """Reset device configurations which may have been set by tests. Should be
105    done before tests start (in case previous one was killed without tearing
106    down) and after they end (to leave device in usable state).
107
108    Args:
109      ad: device to be reset
110    """
111        ad.adb.shell("cmd wifiaware reset")
112
113    def reset_device_statistics(self, ad):
114        """Reset device statistics.
115
116    Args:
117        ad: device to be reset
118    """
119        ad.adb.shell("cmd wifiaware native_cb get_cb_count --reset")
120
121    def set_power_mode_parameters(self, ad):
122        """Set the power configuration DW parameters for the device based on any
123    configuration overrides (if provided)"""
124        if self.aware_default_power_mode == "INTERACTIVE":
125            autils.config_settings_high_power(ad)
126        elif self.aware_default_power_mode == "NON_INTERACTIVE":
127            autils.config_settings_low_power(ad)
128        else:
129            asserts.assert_false(
130                "The 'aware_default_power_mode' configuration must be INTERACTIVE or "
131                "NON_INTERACTIVE")
132
133    def get_next_msg_id(self):
134        """Increment the message ID and returns the new value. Guarantees that
135    each call to the method returns a unique value.
136
137    Returns: a new message id value.
138    """
139        self.msg_id = self.msg_id + 1
140        return self.msg_id
141
142    def on_fail(self, test_name, begin_time):
143        for ad in self.android_devices:
144            ad.take_bug_report(test_name, begin_time)
145            ad.cat_adb_log(test_name, begin_time)
146            wutils.get_ssrdumps(ad)
147        if hasattr(self, "cnss_diag_file") and hasattr(self, "pixel_models"):
148            wutils.stop_cnss_diags(self.android_devices, self.pixel_models)
149            for ad in self.android_devices:
150                wutils.get_cnss_diag_log(ad)
151        for proc in self.tcpdump_proc:
152            nutils.stop_tcpdump(proc[0], proc[1], self.test_name)
153        self.tcpdump_proc = []
154