1#!/usr/bin/python3.4
2#
3#   Copyright 2017 - 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 queue
18import time
19
20from acts import asserts
21from acts import utils
22from acts.test_decorators import test_tracker_info
23from acts.test_utils.wifi import wifi_test_utils as wutils
24from acts.test_utils.wifi import wifi_constants as wconsts
25from acts.test_utils.wifi.aware import aware_const as aconsts
26from acts.test_utils.wifi.aware import aware_test_utils as autils
27from acts.test_utils.wifi.aware.AwareBaseTest import AwareBaseTest
28
29# arbitrary timeout for events
30EVENT_TIMEOUT = 10
31
32
33class NonConcurrencyTest(AwareBaseTest):
34    """Tests lack of concurrency scenarios Wi-Fi Aware with WFD (p2p) and
35  SoftAP
36
37  Note: these tests should be modified if the concurrency behavior changes!"""
38
39    SERVICE_NAME = "GoogleTestXYZ"
40    TETHER_SSID = "GoogleTestSoftApXYZ"
41
42    def teardown_test(self):
43        AwareBaseTest.teardown_test(self)
44        for ad in self.android_devices:
45            ad.droid.wifiP2pClose()
46            ad.droid.connectivityStopTethering(0)
47
48    def run_aware_then_incompat_service(self, is_p2p):
49        """Run test to validate that a running Aware session terminates when an
50    Aware-incompatible service is started.
51    P2P: has same priority, will bring down Aware, then Aware will become available again.
52    SoftAp: has higher priority, will bring down Aware, Aware will keep unavailable.
53
54    Args:
55      is_p2p: True for p2p, False for SoftAP
56    """
57        dut = self.android_devices[0]
58
59        # start Aware
60        id = dut.droid.wifiAwareAttach()
61        autils.wait_for_event(dut, aconsts.EVENT_CB_ON_ATTACHED)
62
63        time.sleep(EVENT_TIMEOUT)
64
65        # start other service
66        if is_p2p:
67            dut.droid.wifiP2pInitialize()
68        else:
69            wutils.start_wifi_tethering(dut, self.TETHER_SSID, password=None)
70
71        # expect an announcement about Aware non-availability
72        autils.wait_for_event(dut, aconsts.BROADCAST_WIFI_AWARE_NOT_AVAILABLE)
73
74        if is_p2p:
75            # P2P has same priority, aware will be available
76            autils.wait_for_event(dut, aconsts.BROADCAST_WIFI_AWARE_AVAILABLE)
77        else:
78            # SoftAp has higher priority, aware will keep unavailable
79            autils.fail_on_event(dut, aconsts.BROADCAST_WIFI_AWARE_AVAILABLE)
80            # local clean-up
81            wutils.stop_wifi_tethering(dut)
82
83    def run_incompat_service_then_aware(self, is_p2p):
84        """Validate that if an Aware-incompatible service is already up then try to start Aware.
85    P2P: has same priority, Aware can bring it down.
86    SoftAp: has higher priority, Aware will be unavailable, any Aware operation will fail.
87
88    Args:
89      is_p2p: True for p2p, False for SoftAP
90    """
91        dut = self.android_devices[0]
92
93        # start other service
94        if is_p2p:
95            dut.droid.wifiP2pInitialize()
96            # expect no announcement about Aware non-availability
97            autils.fail_on_event(dut, aconsts.BROADCAST_WIFI_AWARE_NOT_AVAILABLE)
98        else:
99            wutils.start_wifi_tethering(dut, self.TETHER_SSID, password=None)
100            # expect an announcement about Aware non-availability
101            autils.wait_for_event(dut, aconsts.BROADCAST_WIFI_AWARE_NOT_AVAILABLE)
102
103        # Change Wifi state and location mode to check if aware became available.
104        wutils.wifi_toggle_state(dut, False)
105        utils.set_location_service(dut, False)
106        wutils.wifi_toggle_state(dut, True)
107        utils.set_location_service(dut, True)
108
109        if is_p2p:
110            # P2P has same priority, aware will be available
111            autils.wait_for_event(dut, aconsts.BROADCAST_WIFI_AWARE_AVAILABLE)
112            asserts.assert_true(dut.droid.wifiIsAwareAvailable(), "Aware should be available")
113        else:
114            # SoftAp has higher priority, aware will keep unavailable
115            autils.fail_on_event(dut, aconsts.BROADCAST_WIFI_AWARE_AVAILABLE)
116            asserts.assert_false(dut.droid.wifiIsAwareAvailable(),
117                                 "Aware is available (it shouldn't be)")
118
119        dut.droid.wifiAwareAttach()
120        if is_p2p:
121            # P2P has same priority, Aware attach should success.
122            autils.wait_for_event(dut, aconsts.EVENT_CB_ON_ATTACHED)
123        else:
124            # SoftAp has higher priority, Aware attach should fail.
125            autils.wait_for_event(dut, aconsts.EVENT_CB_ON_ATTACH_FAILED)
126
127        if not is_p2p:
128            wutils.stop_wifi_tethering(dut)
129
130            # expect an announcement about Aware availability
131            autils.wait_for_event(dut, aconsts.BROADCAST_WIFI_AWARE_AVAILABLE)
132
133            # try starting Aware
134            dut.droid.wifiAwareAttach()
135            autils.wait_for_event(dut, aconsts.EVENT_CB_ON_ATTACHED)
136
137    def run_aware_then_connect_to_new_ap(self):
138        """Validate interaction of Wi-Fi Aware and infra (STA) association with randomized MAC
139    address. Such an association may trigger interface down and up - possibly disrupting a Wi-Fi
140    Aware session.
141
142    Test behavior:
143    - Start Aware
144    - Associate STA
145    - Check if an Aware state change Broadcast received
146    - If necessary (Broadcast received) restart Aware
147    - Start publish
148    - Start Subscribe on peer
149    - Verify discovery
150    """
151        dut = self.android_devices[0]
152        dut_ap = self.android_devices[1]
153        wutils.reset_wifi(dut)
154        wutils.reset_wifi(dut_ap)
155        p_config = autils.create_discovery_config(self.SERVICE_NAME,
156                                                  aconsts.PUBLISH_TYPE_UNSOLICITED)
157        s_config = autils.create_discovery_config(self.SERVICE_NAME,
158                                                  aconsts.SUBSCRIBE_TYPE_PASSIVE)
159
160        # create random SSID and start softAp on dut_ap
161        ap_ssid = self.TETHER_SSID + utils.rand_ascii_str(8)
162        ap_password = utils.rand_ascii_str(8)
163        config = {wutils.WifiEnums.SSID_KEY: ap_ssid, wutils.WifiEnums.PWD_KEY: ap_password}
164        wutils.start_wifi_tethering(dut_ap, ap_ssid, ap_password)
165        asserts.assert_true(dut_ap.droid.wifiIsApEnabled(),
166                            "SoftAp is not reported as running")
167
168        # dut start Aware attach and connect to softAp on dut_ap
169        p_id = dut.droid.wifiAwareAttach()
170        autils.wait_for_event(dut, aconsts.EVENT_CB_ON_ATTACHED)
171
172        wutils.start_wifi_connection_scan_and_ensure_network_found(dut, ap_ssid)
173        wutils.wifi_connect(dut, config, check_connectivity=False)
174        autils.wait_for_event(dut, wconsts.WIFI_STATE_CHANGED)
175
176        # Check if the WifiAwareState changes then restart the Aware
177        try:
178            dut.ed.pop_event(aconsts.BROADCAST_WIFI_AWARE_AVAILABLE, EVENT_TIMEOUT)
179            dut.log.info(aconsts.BROADCAST_WIFI_AWARE_AVAILABLE)
180            p_id = dut.droid.wifiAwareAttach()
181            autils.wait_for_event(dut, aconsts.EVENT_CB_ON_ATTACHED)
182        except queue.Empty:
183            dut.log.info('WifiAware state was not changed')
184
185        # dut start Publish
186        p_disc_id = dut.droid.wifiAwarePublish(p_id, p_config)
187        autils.wait_for_event(dut, aconsts.SESSION_CB_ON_PUBLISH_STARTED)
188
189        # dut_ap stop softAp and start Subscribe
190        wutils.stop_wifi_tethering(dut_ap)
191        autils.wait_for_event(dut_ap, aconsts.BROADCAST_WIFI_AWARE_AVAILABLE)
192        s_id = dut_ap.droid.wifiAwareAttach()
193        autils.wait_for_event(dut_ap, aconsts.EVENT_CB_ON_ATTACHED)
194        s_disc_id = dut_ap.droid.wifiAwareSubscribe(s_id, s_config)
195        autils.wait_for_event(dut_ap, aconsts.SESSION_CB_ON_SUBSCRIBE_STARTED)
196
197        # Check discovery session
198        autils.wait_for_event(dut_ap, aconsts.SESSION_CB_ON_SERVICE_DISCOVERED)
199
200    ##########################################################################
201
202    @test_tracker_info(uuid="b7c84cbe-d744-440a-9279-a0133e88e8cb")
203    def test_run_p2p_then_aware(self):
204        """Validate that if p2p is already up then any Aware operation fails"""
205        self.run_incompat_service_then_aware(is_p2p=True)
206
207    @test_tracker_info(uuid="1e7b3a6d-575d-4911-80bb-6fcf1157ee9f")
208    def test_run_aware_then_p2p(self):
209        """Validate that a running Aware session terminates when p2p is started"""
210        self.run_aware_then_incompat_service(is_p2p=True)
211
212    @test_tracker_info(uuid="82a0bd98-3022-4831-ac9e-d81f58c742d2")
213    def test_run_softap_then_aware(self):
214        """Validate that if SoftAp is already up then any Aware operation fails"""
215        asserts.skip_if(
216            self.android_devices[0].model not in self.dbs_supported_models,
217            "Device %s doesn't support STA+AP." % self.android_devices[0].model)
218        self.run_incompat_service_then_aware(is_p2p=False)
219
220    @test_tracker_info(uuid="0da7661e-8ac2-4f68-b6d3-b3f612369d03")
221    def test_run_aware_then_softap(self):
222        """Validate that a running Aware session terminates when softAp is
223    started"""
224        self.run_aware_then_incompat_service(is_p2p=False)
225
226    @test_tracker_info(uuid="2ac27ac6-8010-4d05-b892-00242420b075")
227    def test_run_aware_then_connect_new_ap(self):
228        """Validate connect new ap during Aware session"""
229        self.run_aware_then_connect_to_new_ap()
230