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 18 19from acts.test_decorators import test_tracker_info 20from acts.test_utils.net import connectivity_const as cconsts 21from acts.test_utils.wifi.aware import aware_const as aconsts 22from acts.test_utils.wifi.aware import aware_test_utils as autils 23from acts.test_utils.wifi.aware.AwareBaseTest import AwareBaseTest 24 25 26class CapabilitiesTest(AwareBaseTest): 27 """Set of tests for Wi-Fi Aware Capabilities - verifying that the provided 28 capabilities are real (i.e. available).""" 29 30 SERVICE_NAME = "GoogleTestXYZ" 31 32 def create_config(self, dtype, service_name): 33 """Create a discovery configuration based on input parameters. 34 35 Args: 36 dtype: Publish or Subscribe discovery type 37 service_name: Service name. 38 39 Returns: 40 Discovery configuration object. 41 """ 42 config = {} 43 config[aconsts.DISCOVERY_KEY_DISCOVERY_TYPE] = dtype 44 config[aconsts.DISCOVERY_KEY_SERVICE_NAME] = service_name 45 return config 46 47 def start_discovery_session(self, dut, session_id, is_publish, dtype, 48 service_name, expect_success): 49 """Start a discovery session 50 51 Args: 52 dut: Device under test 53 session_id: ID of the Aware session in which to start discovery 54 is_publish: True for a publish session, False for subscribe session 55 dtype: Type of the discovery session 56 service_name: Service name to use for the discovery session 57 expect_success: True if expect session to be created, False otherwise 58 59 Returns: 60 Discovery session ID. 61 """ 62 config = {} 63 config[aconsts.DISCOVERY_KEY_DISCOVERY_TYPE] = dtype 64 config[aconsts.DISCOVERY_KEY_SERVICE_NAME] = service_name 65 66 if is_publish: 67 disc_id = dut.droid.wifiAwarePublish(session_id, config) 68 event_name = aconsts.SESSION_CB_ON_PUBLISH_STARTED 69 else: 70 disc_id = dut.droid.wifiAwareSubscribe(session_id, config) 71 event_name = aconsts.SESSION_CB_ON_SUBSCRIBE_STARTED 72 73 if expect_success: 74 autils.wait_for_event(dut, event_name) 75 else: 76 autils.wait_for_event(dut, 77 aconsts.SESSION_CB_ON_SESSION_CONFIG_FAILED) 78 79 return disc_id 80 81 ############################### 82 83 @test_tracker_info(uuid="45da8a41-6c02-4434-9eb9-aa0a36ff9f65") 84 def test_max_discovery_sessions(self): 85 """Validate that the device can create as many discovery sessions as are 86 indicated in the device capabilities 87 """ 88 dut = self.android_devices[0] 89 90 # attach 91 session_id = dut.droid.wifiAwareAttach(True) 92 autils.wait_for_event(dut, aconsts.EVENT_CB_ON_ATTACHED) 93 94 service_name_template = 'GoogleTestService-%s-%d' 95 96 # start the max number of publish sessions 97 for i in range(dut.aware_capabilities[aconsts.CAP_MAX_PUBLISHES]): 98 # create publish discovery session of both types 99 pub_disc_id = self.start_discovery_session( 100 dut, session_id, True, aconsts.PUBLISH_TYPE_UNSOLICITED 101 if i % 2 == 0 else aconsts.PUBLISH_TYPE_SOLICITED, 102 service_name_template % ('pub', i), True) 103 104 # start the max number of subscribe sessions 105 for i in range(dut.aware_capabilities[aconsts.CAP_MAX_SUBSCRIBES]): 106 # create publish discovery session of both types 107 sub_disc_id = self.start_discovery_session( 108 dut, session_id, False, aconsts.SUBSCRIBE_TYPE_PASSIVE 109 if i % 2 == 0 else aconsts.SUBSCRIBE_TYPE_ACTIVE, 110 service_name_template % ('sub', i), True) 111 112 # start another publish & subscribe and expect failure 113 self.start_discovery_session( 114 dut, session_id, True, aconsts.PUBLISH_TYPE_UNSOLICITED, 115 service_name_template % ('pub', 900), False) 116 self.start_discovery_session( 117 dut, session_id, False, aconsts.SUBSCRIBE_TYPE_ACTIVE, 118 service_name_template % ('pub', 901), False) 119 120 # delete one of the publishes and try again (see if can create subscribe 121 # instead - should not) 122 dut.droid.wifiAwareDestroyDiscoverySession(pub_disc_id) 123 self.start_discovery_session( 124 dut, session_id, False, aconsts.SUBSCRIBE_TYPE_ACTIVE, 125 service_name_template % ('pub', 902), False) 126 self.start_discovery_session( 127 dut, session_id, True, aconsts.PUBLISH_TYPE_UNSOLICITED, 128 service_name_template % ('pub', 903), True) 129 130 # delete one of the subscribes and try again (see if can create publish 131 # instead - should not) 132 dut.droid.wifiAwareDestroyDiscoverySession(sub_disc_id) 133 self.start_discovery_session( 134 dut, session_id, True, aconsts.PUBLISH_TYPE_UNSOLICITED, 135 service_name_template % ('pub', 904), False) 136 self.start_discovery_session( 137 dut, session_id, False, aconsts.SUBSCRIBE_TYPE_ACTIVE, 138 service_name_template % ('pub', 905), True) 139 140 def test_max_ndp(self): 141 """Validate that the device can create as many NDPs as are specified 142 by its capabilities. 143 144 Mechanics: 145 - Publisher on DUT (first device) 146 - Subscribers on all other devices 147 - On discovery set up NDP 148 149 Note: the test requires MAX_NDP + 2 devices to be validated. If these are 150 not available the test will fail. 151 """ 152 dut = self.android_devices[0] 153 154 # get max NDP: using first available device (assumes all devices are the 155 # same) 156 max_ndp = dut.aware_capabilities[aconsts.CAP_MAX_NDP_SESSIONS] 157 158 # get number of attached devices: needs to be max_ndp+2 to allow for max_ndp 159 # NDPs + an additional one expected to fail. 160 # However, will run the test with max_ndp+1 devices to verify that at least 161 # that many NDPs can be created. Will still fail at the end to indicate that 162 # full test was not run. 163 num_peer_devices = min(len(self.android_devices) - 1, max_ndp + 1) 164 asserts.assert_true( 165 num_peer_devices >= max_ndp, 166 'A minimum of %d devices is needed to run the test, have %d' % 167 (max_ndp + 1, len(self.android_devices))) 168 169 # attach 170 session_id = dut.droid.wifiAwareAttach() 171 autils.wait_for_event(dut, aconsts.EVENT_CB_ON_ATTACHED) 172 173 # start publisher 174 p_disc_id = self.start_discovery_session( 175 dut, 176 session_id, 177 is_publish=True, 178 dtype=aconsts.PUBLISH_TYPE_UNSOLICITED, 179 service_name=self.SERVICE_NAME, 180 expect_success=True) 181 182 # loop over other DUTs 183 for i in range(num_peer_devices): 184 other_dut = self.android_devices[i + 1] 185 186 # attach 187 other_session_id = other_dut.droid.wifiAwareAttach() 188 autils.wait_for_event(other_dut, aconsts.EVENT_CB_ON_ATTACHED) 189 190 # start subscriber 191 s_disc_id = self.start_discovery_session( 192 other_dut, 193 other_session_id, 194 is_publish=False, 195 dtype=aconsts.SUBSCRIBE_TYPE_PASSIVE, 196 service_name=self.SERVICE_NAME, 197 expect_success=True) 198 199 discovery_event = autils.wait_for_event( 200 other_dut, aconsts.SESSION_CB_ON_SERVICE_DISCOVERED) 201 peer_id_on_sub = discovery_event['data'][ 202 aconsts.SESSION_CB_KEY_PEER_ID] 203 204 # Subscriber: send message to peer (Publisher - so it knows our address) 205 other_dut.droid.wifiAwareSendMessage(s_disc_id, peer_id_on_sub, 206 self.get_next_msg_id(), 207 "ping", 208 aconsts.MAX_TX_RETRIES) 209 autils.wait_for_event(other_dut, 210 aconsts.SESSION_CB_ON_MESSAGE_SENT) 211 212 # Publisher: wait for received message 213 pub_rx_msg_event = autils.wait_for_event( 214 dut, aconsts.SESSION_CB_ON_MESSAGE_RECEIVED) 215 peer_id_on_pub = pub_rx_msg_event['data'][ 216 aconsts.SESSION_CB_KEY_PEER_ID] 217 218 # publisher (responder): request network 219 p_req_key = autils.request_network( 220 dut, 221 dut.droid.wifiAwareCreateNetworkSpecifier( 222 p_disc_id, peer_id_on_pub)) 223 224 # subscriber (initiator): request network 225 s_req_key = autils.request_network( 226 other_dut, 227 other_dut.droid.wifiAwareCreateNetworkSpecifier( 228 s_disc_id, peer_id_on_sub)) 229 230 # wait for network (or not - on the last iteration) 231 if i != max_ndp: 232 p_net_event = autils.wait_for_event_with_keys( 233 dut, cconsts.EVENT_NETWORK_CALLBACK, 234 autils.EVENT_NDP_TIMEOUT, 235 (cconsts.NETWORK_CB_KEY_EVENT, 236 cconsts.NETWORK_CB_LINK_PROPERTIES_CHANGED), 237 (cconsts.NETWORK_CB_KEY_ID, p_req_key)) 238 s_net_event = autils.wait_for_event_with_keys( 239 other_dut, cconsts.EVENT_NETWORK_CALLBACK, 240 autils.EVENT_NDP_TIMEOUT, 241 (cconsts.NETWORK_CB_KEY_EVENT, 242 cconsts.NETWORK_CB_LINK_PROPERTIES_CHANGED), 243 (cconsts.NETWORK_CB_KEY_ID, s_req_key)) 244 245 p_aware_if = p_net_event['data'][ 246 cconsts.NETWORK_CB_KEY_INTERFACE_NAME] 247 s_aware_if = s_net_event['data'][ 248 cconsts.NETWORK_CB_KEY_INTERFACE_NAME] 249 self.log.info('Interface names: p=%s, s=%s', p_aware_if, 250 s_aware_if) 251 252 p_ipv6 = dut.droid.connectivityGetLinkLocalIpv6Address( 253 p_aware_if).split('%')[0] 254 s_ipv6 = other_dut.droid.connectivityGetLinkLocalIpv6Address( 255 s_aware_if).split('%')[0] 256 self.log.info('Interface addresses (IPv6): p=%s, s=%s', p_ipv6, 257 s_ipv6) 258 else: 259 autils.fail_on_event_with_keys( 260 dut, cconsts.EVENT_NETWORK_CALLBACK, 261 autils.EVENT_NDP_TIMEOUT, 262 (cconsts.NETWORK_CB_KEY_EVENT, 263 cconsts.NETWORK_CB_LINK_PROPERTIES_CHANGED), 264 (cconsts.NETWORK_CB_KEY_ID, p_req_key)) 265 autils.fail_on_event_with_keys( 266 other_dut, cconsts.EVENT_NETWORK_CALLBACK, 267 autils.EVENT_NDP_TIMEOUT, 268 (cconsts.NETWORK_CB_KEY_EVENT, 269 cconsts.NETWORK_CB_LINK_PROPERTIES_CHANGED), 270 (cconsts.NETWORK_CB_KEY_ID, s_req_key)) 271 272 asserts.assert_true( 273 num_peer_devices > max_ndp, 274 'Needed %d devices to run the test, have %d' % 275 (max_ndp + 2, len(self.android_devices))) 276