#!/usr/bin/python3.4
#
#   Copyright 2017 - The Android Open Source Project
#
#   Licensed under the Apache License, Version 2.0 (the "License");
#   you may not use this file except in compliance with the License.
#   You may obtain a copy of the License at
#
#       http://www.apache.org/licenses/LICENSE-2.0
#
#   Unless required by applicable law or agreed to in writing, software
#   distributed under the License is distributed on an "AS IS" BASIS,
#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#   See the License for the specific language governing permissions and
#   limitations under the License.

from acts import asserts
from acts.test_decorators import test_tracker_info
from acts_contrib.test_utils.net import nsd_const as nconsts
from acts_contrib.test_utils.wifi.aware import aware_const as aconsts
from acts_contrib.test_utils.wifi.aware import aware_test_utils as autils
from acts_contrib.test_utils.wifi.aware.AwareBaseTest import AwareBaseTest


class ProtocolsTest(AwareBaseTest):
    """Set of tests for Wi-Fi Aware data-paths: validating protocols running on
  top of a data-path"""

    SERVICE_NAME = "GoogleTestServiceXY"

    def run_ping6(self, dut, peer_ipv6):
        """Run a ping6 over the specified device/link

    Args:
      dut: Device on which to execute ping6
      peer_ipv6: Scoped IPv6 address of the peer to ping
    """
        cmd = "ping6 -c 3 -W 5 %s" % peer_ipv6
        results = dut.adb.shell(cmd)
        self.log.info("cmd='%s' -> '%s'", cmd, results)
        if results == "":
            asserts.fail("ping6 empty results - seems like a failure")

    ########################################################################

    @test_tracker_info(uuid="ce103067-7fdd-4379-9a2b-d238959f1d53")
    def test_ping6_oob(self):
        """Validate that ping6 works correctly on an NDP created using OOB (out-of
    band) discovery"""
        init_dut = self.android_devices[0]
        resp_dut = self.android_devices[1]

        # create NDP
        (init_req_key, resp_req_key, init_aware_if, resp_aware_if, init_ipv6,
         resp_ipv6) = autils.create_oob_ndp(init_dut, resp_dut)
        self.log.info("Interface names: I=%s, R=%s", init_aware_if,
                      resp_aware_if)
        self.log.info("Interface addresses (IPv6): I=%s, R=%s", init_ipv6,
                      resp_ipv6)

        # run ping6
        self.run_ping6(init_dut, resp_ipv6)
        self.run_ping6(resp_dut, init_ipv6)

        # clean-up
        resp_dut.droid.connectivityUnregisterNetworkCallback(resp_req_key)
        init_dut.droid.connectivityUnregisterNetworkCallback(init_req_key)

    @test_tracker_info(uuid="fef86a48-0e05-464b-8c66-64316275c5ba")
    def test_ping6_ib_unsolicited_passive(self):
        """Validate that ping6 works correctly on an NDP created using Aware
    discovery with UNSOLICITED/PASSIVE sessions."""
        p_dut = self.android_devices[0]
        s_dut = self.android_devices[1]

        # create NDP
        (p_req_key, s_req_key, p_aware_if, s_aware_if, p_ipv6,
         s_ipv6) = autils.create_ib_ndp(
             p_dut,
             s_dut,
             p_config=autils.create_discovery_config(
                 self.SERVICE_NAME, aconsts.PUBLISH_TYPE_UNSOLICITED),
             s_config=autils.create_discovery_config(
                 self.SERVICE_NAME, aconsts.SUBSCRIBE_TYPE_PASSIVE),
             device_startup_offset=self.device_startup_offset)
        self.log.info("Interface names: P=%s, S=%s", p_aware_if, s_aware_if)
        self.log.info("Interface addresses (IPv6): P=%s, S=%s", p_ipv6, s_ipv6)

        # run ping6
        self.run_ping6(p_dut, s_ipv6)
        self.run_ping6(s_dut, p_ipv6)

        # clean-up
        p_dut.droid.connectivityUnregisterNetworkCallback(p_req_key)
        s_dut.droid.connectivityUnregisterNetworkCallback(s_req_key)

    @test_tracker_info(uuid="5bbd68a9-994b-4c26-88cd-43388cec280b")
    def test_ping6_ib_solicited_active(self):
        """Validate that ping6 works correctly on an NDP created using Aware
    discovery with SOLICITED/ACTIVE sessions."""
        p_dut = self.android_devices[0]
        s_dut = self.android_devices[1]

        # create NDP
        (p_req_key, s_req_key, p_aware_if, s_aware_if, p_ipv6,
         s_ipv6) = autils.create_ib_ndp(
             p_dut,
             s_dut,
             p_config=autils.create_discovery_config(
                 self.SERVICE_NAME, aconsts.PUBLISH_TYPE_SOLICITED),
             s_config=autils.create_discovery_config(
                 self.SERVICE_NAME, aconsts.SUBSCRIBE_TYPE_ACTIVE),
             device_startup_offset=self.device_startup_offset)
        self.log.info("Interface names: P=%s, S=%s", p_aware_if, s_aware_if)
        self.log.info("Interface addresses (IPv6): P=%s, S=%s", p_ipv6, s_ipv6)

        # run ping6
        self.run_ping6(p_dut, s_ipv6)
        self.run_ping6(s_dut, p_ipv6)

        # clean-up
        p_dut.droid.connectivityUnregisterNetworkCallback(p_req_key)
        s_dut.droid.connectivityUnregisterNetworkCallback(s_req_key)

    def test_ping6_oob_max_ndp(self):
        """Validate that ping6 works correctly on multiple NDPs brought up
    concurrently. Uses the capability of the device to determine the max
    number of NDPs to set up.

    Note: the test requires MAX_NDP + 1 devices to be validated. If these are
    not available the test will fail."""
        dut = self.android_devices[0]

        # get max NDP: using first available device (assumes all devices are the
        # same)
        max_ndp = dut.aware_capabilities[aconsts.CAP_MAX_NDP_SESSIONS]
        asserts.assert_true(
            len(self.android_devices) > max_ndp,
            'Needed %d devices to run the test, have %d' %
            (max_ndp + 1, len(self.android_devices)))

        # create all NDPs
        dut_aware_if = None
        dut_ipv6 = None
        peers_aware_ifs = []
        peers_ipv6s = []
        dut_requests = []
        peers_requests = []
        for i in range(max_ndp):
            (init_req_key, resp_req_key, init_aware_if, resp_aware_if,
             init_ipv6, resp_ipv6) = autils.create_oob_ndp(
                 dut, self.android_devices[i + 1])
            self.log.info("Interface names: I=%s, R=%s", init_aware_if,
                          resp_aware_if)
            self.log.info("Interface addresses (IPv6): I=%s, R=%s", init_ipv6,
                          resp_ipv6)

            dut_requests.append(init_req_key)
            peers_requests.append(resp_req_key)
            if dut_aware_if is None:
                dut_aware_if = init_aware_if
            else:
                asserts.assert_equal(
                    dut_aware_if, init_aware_if,
                    "DUT (Initiator) interface changed on subsequent NDPs!?")
            if dut_ipv6 is None:
                dut_ipv6 = init_ipv6
            else:
                asserts.assert_equal(
                    dut_ipv6, init_ipv6,
                    "DUT (Initiator) IPv6 changed on subsequent NDPs!?")
            peers_aware_ifs.append(resp_aware_if)
            peers_ipv6s.append(resp_ipv6)

        # run ping6
        for i in range(max_ndp):
            self.run_ping6(dut, peers_ipv6s[i])
            self.run_ping6(self.android_devices[i + 1], dut_ipv6)

        # cleanup
        for i in range(max_ndp):
            dut.droid.connectivityUnregisterNetworkCallback(dut_requests[i])
            self.android_devices[
                i + 1].droid.connectivityUnregisterNetworkCallback(
                    peers_requests[i])

    def test_nsd_oob(self):
        """Validate that NSD (mDNS) works correctly on an NDP created using OOB
    (out-of band) discovery"""
        init_dut = self.android_devices[0]
        resp_dut = self.android_devices[1]

        # create NDP
        (init_req_key, resp_req_key, init_aware_if, resp_aware_if, init_ipv6,
         resp_ipv6) = autils.create_oob_ndp(init_dut, resp_dut)
        self.log.info("Interface names: I=%s, R=%s", init_aware_if,
                      resp_aware_if)
        self.log.info("Interface addresses (IPv6): I=%s, R=%s", init_ipv6,
                      resp_ipv6)

        # run NSD
        nsd_service_info = {
            "serviceInfoServiceName": "sl4aTestAwareNsd",
            "serviceInfoServiceType": "_simple-tx-rx._tcp.",
            "serviceInfoPort": 2257
        }
        nsd_reg = None
        nsd_discovery = None
        try:
            # Initiator registers an NSD service
            nsd_reg = init_dut.droid.nsdRegisterService(nsd_service_info)
            event_nsd = autils.wait_for_event_with_keys(
                init_dut, nconsts.REG_LISTENER_EVENT, autils.EVENT_TIMEOUT,
                (nconsts.REG_LISTENER_CALLBACK,
                 nconsts.REG_LISTENER_EVENT_ON_SERVICE_REGISTERED))
            self.log.info("Initiator %s: %s",
                          nconsts.REG_LISTENER_EVENT_ON_SERVICE_REGISTERED,
                          event_nsd["data"])

            # Responder starts an NSD discovery
            nsd_discovery = resp_dut.droid.nsdDiscoverServices(
                nsd_service_info[nconsts.NSD_SERVICE_INFO_SERVICE_TYPE])
            event_nsd = autils.wait_for_event_with_keys(
                resp_dut, nconsts.DISCOVERY_LISTENER_EVENT,
                autils.EVENT_TIMEOUT,
                (nconsts.DISCOVERY_LISTENER_DATA_CALLBACK,
                 nconsts.DISCOVERY_LISTENER_EVENT_ON_SERVICE_FOUND))
            self.log.info("Responder %s: %s",
                          nconsts.DISCOVERY_LISTENER_EVENT_ON_SERVICE_FOUND,
                          event_nsd["data"])

            # Responder resolves IP address of Initiator from NSD service discovery
            resp_dut.droid.nsdResolveService(event_nsd["data"])
            event_nsd = autils.wait_for_event_with_keys(
                resp_dut, nconsts.RESOLVE_LISTENER_EVENT, autils.EVENT_TIMEOUT,
                (nconsts.RESOLVE_LISTENER_DATA_CALLBACK,
                 nconsts.RESOLVE_LISTENER_EVENT_ON_SERVICE_RESOLVED))
            self.log.info("Responder %s: %s",
                          nconsts.RESOLVE_LISTENER_EVENT_ON_SERVICE_RESOLVED,
                          event_nsd["data"])

            # mDNS returns first character as '/' - strip
            # out to get clean IPv6
            init_ipv6_nsd = event_nsd["data"][nconsts.NSD_SERVICE_INFO_HOST][
                1:]

            # mDNS doesn't seem to return a scoped host so strip it out of the
            # local result (for now)
            scope_index = init_ipv6.find("%")
            if scope_index != -1:
                init_ipv6 = init_ipv6[:scope_index]

            asserts.assert_equal(
                init_ipv6, init_ipv6_nsd,
                "Initiator's IPv6 address obtained through NSD doesn't match!?"
            )
        finally:
            # Stop NSD
            if nsd_reg is not None:
                init_dut.droid.nsdUnregisterService(nsd_reg)
            if nsd_discovery is not None:
                resp_dut.droid.nsdStopServiceDiscovery(nsd_discovery)

        # clean-up
        resp_dut.droid.connectivityUnregisterNetworkCallback(resp_req_key)
        init_dut.droid.connectivityUnregisterNetworkCallback(init_req_key)