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 18from acts.test_decorators import test_tracker_info 19from acts.test_utils.net import nsd_const as nconsts 20from acts.test_utils.wifi.aware import aware_const as aconsts 21from acts.test_utils.wifi.aware import aware_test_utils as autils 22from acts.test_utils.wifi.aware.AwareBaseTest import AwareBaseTest 23 24 25class ProtocolsTest(AwareBaseTest): 26 """Set of tests for Wi-Fi Aware data-paths: validating protocols running on 27 top of a data-path""" 28 29 SERVICE_NAME = "GoogleTestServiceXY" 30 31 def run_ping6(self, dut, peer_ipv6): 32 """Run a ping6 over the specified device/link 33 34 Args: 35 dut: Device on which to execute ping6 36 peer_ipv6: Scoped IPv6 address of the peer to ping 37 """ 38 cmd = "ping6 -c 3 -W 5 %s" % peer_ipv6 39 results = dut.adb.shell(cmd) 40 self.log.info("cmd='%s' -> '%s'", cmd, results) 41 if results == "": 42 asserts.fail("ping6 empty results - seems like a failure") 43 44 ######################################################################## 45 46 @test_tracker_info(uuid="ce103067-7fdd-4379-9a2b-d238959f1d53") 47 def test_ping6_oob(self): 48 """Validate that ping6 works correctly on an NDP created using OOB (out-of 49 band) discovery""" 50 init_dut = self.android_devices[0] 51 resp_dut = self.android_devices[1] 52 53 # create NDP 54 (init_req_key, resp_req_key, init_aware_if, resp_aware_if, init_ipv6, 55 resp_ipv6) = autils.create_oob_ndp(init_dut, resp_dut) 56 self.log.info("Interface names: I=%s, R=%s", init_aware_if, 57 resp_aware_if) 58 self.log.info("Interface addresses (IPv6): I=%s, R=%s", init_ipv6, 59 resp_ipv6) 60 61 # run ping6 62 self.run_ping6(init_dut, resp_ipv6) 63 self.run_ping6(resp_dut, init_ipv6) 64 65 # clean-up 66 resp_dut.droid.connectivityUnregisterNetworkCallback(resp_req_key) 67 init_dut.droid.connectivityUnregisterNetworkCallback(init_req_key) 68 69 @test_tracker_info(uuid="fef86a48-0e05-464b-8c66-64316275c5ba") 70 def test_ping6_ib_unsolicited_passive(self): 71 """Validate that ping6 works correctly on an NDP created using Aware 72 discovery with UNSOLICITED/PASSIVE sessions.""" 73 p_dut = self.android_devices[0] 74 s_dut = self.android_devices[1] 75 76 # create NDP 77 (p_req_key, s_req_key, p_aware_if, s_aware_if, p_ipv6, 78 s_ipv6) = autils.create_ib_ndp( 79 p_dut, 80 s_dut, 81 p_config=autils.create_discovery_config( 82 self.SERVICE_NAME, aconsts.PUBLISH_TYPE_UNSOLICITED), 83 s_config=autils.create_discovery_config( 84 self.SERVICE_NAME, aconsts.SUBSCRIBE_TYPE_PASSIVE), 85 device_startup_offset=self.device_startup_offset) 86 self.log.info("Interface names: P=%s, S=%s", p_aware_if, s_aware_if) 87 self.log.info("Interface addresses (IPv6): P=%s, S=%s", p_ipv6, s_ipv6) 88 89 # run ping6 90 self.run_ping6(p_dut, s_ipv6) 91 self.run_ping6(s_dut, p_ipv6) 92 93 # clean-up 94 p_dut.droid.connectivityUnregisterNetworkCallback(p_req_key) 95 s_dut.droid.connectivityUnregisterNetworkCallback(s_req_key) 96 97 @test_tracker_info(uuid="5bbd68a9-994b-4c26-88cd-43388cec280b") 98 def test_ping6_ib_solicited_active(self): 99 """Validate that ping6 works correctly on an NDP created using Aware 100 discovery with SOLICITED/ACTIVE sessions.""" 101 p_dut = self.android_devices[0] 102 s_dut = self.android_devices[1] 103 104 # create NDP 105 (p_req_key, s_req_key, p_aware_if, s_aware_if, p_ipv6, 106 s_ipv6) = autils.create_ib_ndp( 107 p_dut, 108 s_dut, 109 p_config=autils.create_discovery_config( 110 self.SERVICE_NAME, aconsts.PUBLISH_TYPE_SOLICITED), 111 s_config=autils.create_discovery_config( 112 self.SERVICE_NAME, aconsts.SUBSCRIBE_TYPE_ACTIVE), 113 device_startup_offset=self.device_startup_offset) 114 self.log.info("Interface names: P=%s, S=%s", p_aware_if, s_aware_if) 115 self.log.info("Interface addresses (IPv6): P=%s, S=%s", p_ipv6, s_ipv6) 116 117 # run ping6 118 self.run_ping6(p_dut, s_ipv6) 119 self.run_ping6(s_dut, p_ipv6) 120 121 # clean-up 122 p_dut.droid.connectivityUnregisterNetworkCallback(p_req_key) 123 s_dut.droid.connectivityUnregisterNetworkCallback(s_req_key) 124 125 def test_ping6_oob_max_ndp(self): 126 """Validate that ping6 works correctly on multiple NDPs brought up 127 concurrently. Uses the capability of the device to determine the max 128 number of NDPs to set up. 129 130 Note: the test requires MAX_NDP + 1 devices to be validated. If these are 131 not available the test will fail.""" 132 dut = self.android_devices[0] 133 134 # get max NDP: using first available device (assumes all devices are the 135 # same) 136 max_ndp = dut.aware_capabilities[aconsts.CAP_MAX_NDP_SESSIONS] 137 asserts.assert_true( 138 len(self.android_devices) > max_ndp, 139 'Needed %d devices to run the test, have %d' % 140 (max_ndp + 1, len(self.android_devices))) 141 142 # create all NDPs 143 dut_aware_if = None 144 dut_ipv6 = None 145 peers_aware_ifs = [] 146 peers_ipv6s = [] 147 dut_requests = [] 148 peers_requests = [] 149 for i in range(max_ndp): 150 (init_req_key, resp_req_key, init_aware_if, resp_aware_if, 151 init_ipv6, resp_ipv6) = autils.create_oob_ndp( 152 dut, self.android_devices[i + 1]) 153 self.log.info("Interface names: I=%s, R=%s", init_aware_if, 154 resp_aware_if) 155 self.log.info("Interface addresses (IPv6): I=%s, R=%s", init_ipv6, 156 resp_ipv6) 157 158 dut_requests.append(init_req_key) 159 peers_requests.append(resp_req_key) 160 if dut_aware_if is None: 161 dut_aware_if = init_aware_if 162 else: 163 asserts.assert_equal( 164 dut_aware_if, init_aware_if, 165 "DUT (Initiator) interface changed on subsequent NDPs!?") 166 if dut_ipv6 is None: 167 dut_ipv6 = init_ipv6 168 else: 169 asserts.assert_equal( 170 dut_ipv6, init_ipv6, 171 "DUT (Initiator) IPv6 changed on subsequent NDPs!?") 172 peers_aware_ifs.append(resp_aware_if) 173 peers_ipv6s.append(resp_ipv6) 174 175 # run ping6 176 for i in range(max_ndp): 177 self.run_ping6(dut, peers_ipv6s[i]) 178 self.run_ping6(self.android_devices[i + 1], dut_ipv6) 179 180 # cleanup 181 for i in range(max_ndp): 182 dut.droid.connectivityUnregisterNetworkCallback(dut_requests[i]) 183 self.android_devices[ 184 i + 1].droid.connectivityUnregisterNetworkCallback( 185 peers_requests[i]) 186 187 def test_nsd_oob(self): 188 """Validate that NSD (mDNS) works correctly on an NDP created using OOB 189 (out-of band) discovery""" 190 init_dut = self.android_devices[0] 191 resp_dut = self.android_devices[1] 192 193 # create NDP 194 (init_req_key, resp_req_key, init_aware_if, resp_aware_if, init_ipv6, 195 resp_ipv6) = autils.create_oob_ndp(init_dut, resp_dut) 196 self.log.info("Interface names: I=%s, R=%s", init_aware_if, 197 resp_aware_if) 198 self.log.info("Interface addresses (IPv6): I=%s, R=%s", init_ipv6, 199 resp_ipv6) 200 201 # run NSD 202 nsd_service_info = { 203 "serviceInfoServiceName": "sl4aTestAwareNsd", 204 "serviceInfoServiceType": "_simple-tx-rx._tcp.", 205 "serviceInfoPort": 2257 206 } 207 nsd_reg = None 208 nsd_discovery = None 209 try: 210 # Initiator registers an NSD service 211 nsd_reg = init_dut.droid.nsdRegisterService(nsd_service_info) 212 event_nsd = autils.wait_for_event_with_keys( 213 init_dut, nconsts.REG_LISTENER_EVENT, autils.EVENT_TIMEOUT, 214 (nconsts.REG_LISTENER_CALLBACK, 215 nconsts.REG_LISTENER_EVENT_ON_SERVICE_REGISTERED)) 216 self.log.info("Initiator %s: %s", 217 nconsts.REG_LISTENER_EVENT_ON_SERVICE_REGISTERED, 218 event_nsd["data"]) 219 220 # Responder starts an NSD discovery 221 nsd_discovery = resp_dut.droid.nsdDiscoverServices( 222 nsd_service_info[nconsts.NSD_SERVICE_INFO_SERVICE_TYPE]) 223 event_nsd = autils.wait_for_event_with_keys( 224 resp_dut, nconsts.DISCOVERY_LISTENER_EVENT, 225 autils.EVENT_TIMEOUT, 226 (nconsts.DISCOVERY_LISTENER_DATA_CALLBACK, 227 nconsts.DISCOVERY_LISTENER_EVENT_ON_SERVICE_FOUND)) 228 self.log.info("Responder %s: %s", 229 nconsts.DISCOVERY_LISTENER_EVENT_ON_SERVICE_FOUND, 230 event_nsd["data"]) 231 232 # Responder resolves IP address of Initiator from NSD service discovery 233 resp_dut.droid.nsdResolveService(event_nsd["data"]) 234 event_nsd = autils.wait_for_event_with_keys( 235 resp_dut, nconsts.RESOLVE_LISTENER_EVENT, autils.EVENT_TIMEOUT, 236 (nconsts.RESOLVE_LISTENER_DATA_CALLBACK, 237 nconsts.RESOLVE_LISTENER_EVENT_ON_SERVICE_RESOLVED)) 238 self.log.info("Responder %s: %s", 239 nconsts.RESOLVE_LISTENER_EVENT_ON_SERVICE_RESOLVED, 240 event_nsd["data"]) 241 242 # mDNS returns first character as '/' - strip 243 # out to get clean IPv6 244 init_ipv6_nsd = event_nsd["data"][nconsts.NSD_SERVICE_INFO_HOST][ 245 1:] 246 247 # mDNS doesn't seem to return a scoped host so strip it out of the 248 # local result (for now) 249 scope_index = init_ipv6.find("%") 250 if scope_index != -1: 251 init_ipv6 = init_ipv6[:scope_index] 252 253 asserts.assert_equal( 254 init_ipv6, init_ipv6_nsd, 255 "Initiator's IPv6 address obtained through NSD doesn't match!?" 256 ) 257 finally: 258 # Stop NSD 259 if nsd_reg is not None: 260 init_dut.droid.nsdUnregisterService(nsd_reg) 261 if nsd_discovery is not None: 262 resp_dut.droid.nsdStopServiceDiscovery(nsd_discovery) 263 264 # clean-up 265 resp_dut.droid.connectivityUnregisterNetworkCallback(resp_req_key) 266 init_dut.droid.connectivityUnregisterNetworkCallback(init_req_key) 267