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