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
17from acts import asserts
18import queue
19from acts import utils
20from acts.test_utils.wifi import wifi_test_utils as wutils
21from acts.test_utils.wifi import wifi_constants as wconsts
22from acts.test_utils.wifi.aware import aware_const as aconsts
23from acts.test_utils.wifi.aware import aware_test_utils as autils
24from acts.test_utils.wifi.aware.AwareBaseTest import AwareBaseTest
25
26# arbitrary timeout for events
27EVENT_TIMEOUT = 10
28
29
30class NonConcurrencyTest(AwareBaseTest):
31    """Tests lack of concurrency scenarios Wi-Fi Aware with WFD (p2p) and
32  SoftAP
33
34  Note: these tests should be modified if the concurrency behavior changes!"""
35
36    SERVICE_NAME = "GoogleTestXYZ"
37    TETHER_SSID = "GoogleTestSoftApXYZ"
38
39    def teardown_test(self):
40        AwareBaseTest.teardown_test(self)
41        for ad in self.android_devices:
42            ad.droid.wifiP2pClose()
43
44    def run_aware_then_incompat_service(self, is_p2p):
45        """Run test to validate that a running Aware session terminates when an
46    Aware-incompatible service is started.
47
48    Args:
49      is_p2p: True for p2p, False for SoftAP
50    """
51        dut = self.android_devices[0]
52
53        # start Aware
54        id = dut.droid.wifiAwareAttach()
55        autils.wait_for_event(dut, aconsts.EVENT_CB_ON_ATTACHED)
56
57        # start other service
58        if is_p2p:
59            dut.droid.wifiP2pInitialize()
60        else:
61            wutils.start_wifi_tethering(dut, self.TETHER_SSID, password=None)
62
63        # expect an announcement about Aware non-availability
64        autils.wait_for_event(dut, aconsts.BROADCAST_WIFI_AWARE_NOT_AVAILABLE)
65
66        # local clean-up
67        if not is_p2p:
68            wutils.stop_wifi_tethering(dut)
69
70    def run_incompat_service_then_aware(self, is_p2p):
71        """Validate that if an Aware-incompatible service is already up then any
72    Aware operation fails"""
73        dut = self.android_devices[0]
74
75        # start other service
76        if is_p2p:
77            dut.droid.wifiP2pInitialize()
78        else:
79            wutils.start_wifi_tethering(dut, self.TETHER_SSID, password=None)
80
81        # expect an announcement about Aware non-availability
82        autils.wait_for_event(dut, aconsts.BROADCAST_WIFI_AWARE_NOT_AVAILABLE)
83
84        # try starting anyway (expect failure)
85        dut.droid.wifiAwareAttach()
86        autils.wait_for_event(dut, aconsts.EVENT_CB_ON_ATTACH_FAILED)
87
88        # stop other service
89        if is_p2p:
90            dut.droid.wifiP2pClose()
91        else:
92            wutils.stop_wifi_tethering(dut)
93
94        # expect an announcement about Aware availability
95        autils.wait_for_event(dut, aconsts.BROADCAST_WIFI_AWARE_AVAILABLE)
96
97        # try starting Aware
98        dut.droid.wifiAwareAttach()
99        autils.wait_for_event(dut, aconsts.EVENT_CB_ON_ATTACHED)
100
101    def run_aware_then_connect_to_new_ap(self):
102        """Validate interaction of Wi-Fi Aware and infra (STA) association with randomized MAC
103    address. Such an association may trigger interface down and up - possibly disrupting a Wi-Fi
104    Aware session.
105
106    Test behavior:
107    - Start Aware
108    - Associate STA
109    - Check if an Aware state change Broadcast received
110    - If necessary (Broadcast received) restart Aware
111    - Start publish
112    - Start Subscribe on peer
113    - Verify discovery
114    """
115        dut = self.android_devices[0]
116        dut_ap = self.android_devices[1]
117        wutils.reset_wifi(dut)
118        wutils.reset_wifi(dut_ap)
119        p_config = autils.create_discovery_config(self.SERVICE_NAME,
120                                                  aconsts.PUBLISH_TYPE_UNSOLICITED)
121        s_config = autils.create_discovery_config(self.SERVICE_NAME,
122                                                  aconsts.SUBSCRIBE_TYPE_PASSIVE)
123
124        # create random SSID and start softAp on dut_ap
125        ap_ssid = self.TETHER_SSID + utils.rand_ascii_str(8)
126        ap_password = utils.rand_ascii_str(8)
127        config = {wutils.WifiEnums.SSID_KEY: ap_ssid, wutils.WifiEnums.PWD_KEY: ap_password}
128        wutils.start_wifi_tethering(dut_ap, ap_ssid, ap_password)
129        asserts.assert_true(dut_ap.droid.wifiIsApEnabled(),
130                            "SoftAp is not reported as running")
131
132        # dut start Aware attach and connect to softAp on dut_ap
133        p_id = dut.droid.wifiAwareAttach()
134        autils.wait_for_event(dut, aconsts.EVENT_CB_ON_ATTACHED)
135
136        wutils.wifi_connect(dut, config, check_connectivity=False)
137        autils.wait_for_event(dut, wconsts.WIFI_STATE_CHANGED)
138
139        # Check if the WifiAwareState changes then restart the Aware
140        try:
141            dut.ed.pop_event(aconsts.BROADCAST_WIFI_AWARE_AVAILABLE, EVENT_TIMEOUT)
142            dut.log.info(aconsts.BROADCAST_WIFI_AWARE_AVAILABLE)
143            p_id = dut.droid.wifiAwareAttach()
144            autils.wait_for_event(dut, aconsts.EVENT_CB_ON_ATTACHED)
145        except queue.Empty:
146            dut.log.info('WifiAware state was not changed')
147
148        # dut start Publish
149        p_disc_id = dut.droid.wifiAwarePublish(p_id, p_config)
150        autils.wait_for_event(dut, aconsts.SESSION_CB_ON_PUBLISH_STARTED)
151
152        # dut_ap stop softAp and start Subscribe
153        wutils.stop_wifi_tethering(dut_ap)
154        autils.wait_for_event(dut_ap, aconsts.BROADCAST_WIFI_AWARE_AVAILABLE)
155        s_id = dut_ap.droid.wifiAwareAttach()
156        autils.wait_for_event(dut_ap, aconsts.EVENT_CB_ON_ATTACHED)
157        s_disc_id = dut_ap.droid.wifiAwareSubscribe(s_id, s_config)
158        autils.wait_for_event(dut_ap, aconsts.SESSION_CB_ON_SUBSCRIBE_STARTED)
159
160        # Check discovery session
161        autils.wait_for_event(dut_ap, aconsts.SESSION_CB_ON_SERVICE_DISCOVERED)
162
163    ##########################################################################
164
165    def test_run_p2p_then_aware(self):
166        """Validate that if p2p is already up then any Aware operation fails"""
167        self.run_incompat_service_then_aware(is_p2p=True)
168
169    def test_run_aware_then_p2p(self):
170        """Validate that a running Aware session terminates when p2p is started"""
171        self.run_aware_then_incompat_service(is_p2p=True)
172
173    def test_run_softap_then_aware(self):
174        """Validate that if SoftAp is already up then any Aware operation fails"""
175        self.run_incompat_service_then_aware(is_p2p=False)
176
177    def test_run_aware_then_softap(self):
178        """Validate that a running Aware session terminates when softAp is
179    started"""
180        self.run_aware_then_incompat_service(is_p2p=False)
181
182    def test_run_aware_then_connect_new_ap(self):
183        """Validate connect new ap during Aware session"""
184        self.run_aware_then_connect_to_new_ap()
185