1#!/usr/bin/env python3
2#
3#   Copyright 2018 - 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 acts.utils
18import os
19import re
20import time
21
22from acts import asserts
23from acts import utils
24from acts.base_test import BaseTestClass
25from acts.keys import Config
26from acts_contrib.test_utils.net import net_test_utils as nutils
27from acts_contrib.test_utils.wifi import wifi_test_utils as wutils
28from acts_contrib.test_utils.wifi.p2p import wifi_p2p_const as p2pconsts
29
30WAIT_TIME = 60
31
32
33class WifiP2pBaseTest(BaseTestClass):
34    def __init__(self, controllers):
35        if not hasattr(self, 'android_devices'):
36            super(WifiP2pBaseTest, self).__init__(controllers)
37
38    def setup_class(self):
39        for ad in self.android_devices:
40            ad.droid.wakeLockAcquireBright()
41            ad.droid.wakeUpNow()
42        required_params = ()
43        optional_params = ("skip_read_factory_mac", "pixel_models", "cnss_diag_file")
44        self.unpack_userparams(required_params,
45                               optional_params,
46                               skip_read_factory_mac=0)
47
48        self.dut1 = self.android_devices[0]
49        self.dut2 = self.android_devices[1]
50        if self.skip_read_factory_mac:
51            self.dut1_mac = None
52            self.dut2_mac = None
53        else:
54            self.dut1_mac = self.get_p2p_mac_address(self.dut1)
55            self.dut2_mac = self.get_p2p_mac_address(self.dut2)
56
57        #init location before init p2p
58        acts.utils.set_location_service(self.dut1, True)
59        acts.utils.set_location_service(self.dut2, True)
60
61        wutils.wifi_test_device_init(self.dut1)
62        utils.sync_device_time(self.dut1)
63        self.dut1.droid.wifiP2pInitialize()
64        time.sleep(p2pconsts.DEFAULT_FUNCTION_SWITCH_TIME)
65        asserts.assert_true(self.dut1.droid.wifiP2pIsEnabled(),
66                            "DUT1's p2p should be initialized but it didn't")
67        self.dut1.name = "Android_" + self.dut1.serial
68        self.dut1.droid.wifiP2pSetDeviceName(self.dut1.name)
69        wutils.wifi_test_device_init(self.dut2)
70        utils.sync_device_time(self.dut2)
71        self.dut2.droid.wifiP2pInitialize()
72        time.sleep(p2pconsts.DEFAULT_FUNCTION_SWITCH_TIME)
73        asserts.assert_true(self.dut2.droid.wifiP2pIsEnabled(),
74                            "DUT2's p2p should be initialized but it didn't")
75        self.dut2.name = "Android_" + self.dut2.serial
76        self.dut2.droid.wifiP2pSetDeviceName(self.dut2.name)
77
78        if len(self.android_devices) > 2:
79            self.dut3 = self.android_devices[2]
80            acts.utils.set_location_service(self.dut3, True)
81            wutils.wifi_test_device_init(self.dut3)
82            utils.sync_device_time(self.dut3)
83            self.dut3.droid.wifiP2pInitialize()
84            time.sleep(p2pconsts.DEFAULT_FUNCTION_SWITCH_TIME)
85            asserts.assert_true(
86                self.dut3.droid.wifiP2pIsEnabled(),
87                "DUT3's p2p should be initialized but it didn't")
88            self.dut3.name = "Android_" + self.dut3.serial
89            self.dut3.droid.wifiP2pSetDeviceName(self.dut3.name)
90        if hasattr(self, "cnss_diag_file"):
91            if isinstance(self.cnss_diag_file, list):
92                self.cnss_diag_file = self.cnss_diag_file[0]
93            if not os.path.isfile(self.cnss_diag_file):
94                self.cnss_diag_file = os.path.join(
95                    self.user_params[Config.key_config_path.value],
96                    self.cnss_diag_file)
97
98    def teardown_class(self):
99        self.dut1.droid.wifiP2pClose()
100        self.dut2.droid.wifiP2pClose()
101        acts.utils.set_location_service(self.dut1, False)
102        acts.utils.set_location_service(self.dut2, False)
103
104        if len(self.android_devices) > 2:
105            self.dut3.droid.wifiP2pClose()
106            acts.utils.set_location_service(self.dut3, False)
107        for ad in self.android_devices:
108            ad.droid.wakeLockRelease()
109            ad.droid.goToSleepNow()
110
111    def setup_test(self):
112        if hasattr(self, "cnss_diag_file") and hasattr(self, "pixel_models"):
113            wutils.start_cnss_diags(
114                self.android_devices, self.cnss_diag_file, self.pixel_models)
115        self.tcpdump_proc = []
116        if hasattr(self, "android_devices"):
117            for ad in self.android_devices:
118                proc = nutils.start_tcpdump(ad, self.test_name)
119                self.tcpdump_proc.append((ad, proc))
120
121        for ad in self.android_devices:
122            ad.ed.clear_all_events()
123
124    def teardown_test(self):
125        if hasattr(self, "cnss_diag_file") and hasattr(self, "pixel_models"):
126            wutils.stop_cnss_diags(self.android_devices, self.pixel_models)
127        for proc in self.tcpdump_proc:
128            nutils.stop_tcpdump(
129                    proc[0], proc[1], self.test_name, pull_dump=False)
130        self.tcpdump_proc = []
131        for ad in self.android_devices:
132            # Clear p2p group info
133            ad.droid.wifiP2pRequestPersistentGroupInfo()
134            event = ad.ed.pop_event("WifiP2pOnPersistentGroupInfoAvailable",
135                                    p2pconsts.DEFAULT_TIMEOUT)
136            for network in event['data']:
137                ad.droid.wifiP2pDeletePersistentGroup(network['NetworkId'])
138            # Clear p2p local service
139            ad.droid.wifiP2pClearLocalServices()
140
141    def on_fail(self, test_name, begin_time):
142        for ad in self.android_devices:
143            ad.take_bug_report(test_name, begin_time)
144            ad.cat_adb_log(test_name, begin_time)
145            wutils.get_ssrdumps(ad)
146        if hasattr(self, "cnss_diag_file") and hasattr(self, "pixel_models"):
147            wutils.stop_cnss_diags(self.android_devices, self.pixel_models)
148            for ad in self.android_devices:
149                wutils.get_cnss_diag_log(ad)
150        for proc in self.tcpdump_proc:
151            nutils.stop_tcpdump(proc[0], proc[1], self.test_name)
152        self.tcpdump_proc = []
153
154    def get_p2p_mac_address(self, dut):
155        """Gets the current MAC address being used for Wi-Fi Direct."""
156        dut.reboot()
157        time.sleep(WAIT_TIME)
158        out = dut.adb.shell("ifconfig p2p0")
159        return re.match(".* HWaddr (\S+).*", out, re.S).group(1)
160