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