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