acts_tests/tests/google/wifi/p2p/functional/WifiP2pMultiPeersTest.py

#!/usr/bin/env python3
#
#   Copyright 2020 - 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.

import acts_contrib.test_utils.wifi.wifi_test_utils as wutils
import acts.utils
import time

from acts import asserts
from acts import utils

from acts.test_decorators import test_tracker_info
from acts_contrib.test_utils.wifi import wifi_test_utils as wutils
from acts_contrib.test_utils.wifi.p2p.WifiP2pBaseTest import WifiP2pBaseTest
from acts_contrib.test_utils.wifi.p2p import wifi_p2p_test_utils as wp2putils
from acts_contrib.test_utils.wifi.p2p import wifi_p2p_const as p2pconsts

WPS_PBC = wp2putils.WifiP2PEnums.WpsInfo.WIFI_WPS_INFO_PBC
WPS_DISPLAY = wp2putils.WifiP2PEnums.WpsInfo.WIFI_WPS_INFO_DISPLAY
WPS_KEYPAD = wp2putils.WifiP2PEnums.WpsInfo.WIFI_WPS_INFO_KEYPAD

WifiEnums = wutils.WifiEnums


class WifiP2pMultiPeersTest(WifiP2pBaseTest):
    """Tests for multiple clients.

    Test Bed Requirement:
    * 3 Android devices for each test in this class.
    """
    def __init__(self, controllers):
        WifiP2pBaseTest.__init__(self, controllers)

    def setup_test(self):
        WifiP2pBaseTest.setup_test(self)
        asserts.skip_if(
            len(self.android_devices) < 3,
            "No enough android devices. Skip this test")

    def form_group(self, dut1, dut2, isReconnect=False, wpsType=WPS_PBC):
        # Request the connection to create a group
        wp2putils.p2p_connect(dut1, dut2, isReconnect, wpsType)

        if wp2putils.is_go(dut1):
            go_dut = dut1
            gc_dut = dut2
        elif wp2putils.is_go(dut2):
            go_dut = dut2
            gc_dut = dut1
        return (go_dut, gc_dut)

    def verify_group_connection(self, group_clients):
        for gc in group_clients:
            go_ip = wp2putils.p2p_go_ip(gc)
            wp2putils.p2p_connection_ping_test(gc, go_ip)

    def clear_all_events(self, duts):
        for dut in duts:
            dut.ed.clear_all_events()

    def check_disconnection(self, duts):
        for dut in duts:
            wp2putils.check_disconnect(dut)

    """Test Cases"""
    @test_tracker_info(uuid="20cd4f4d-fe7d-4ee2-a832-33caa5b9700b")
    def test_p2p_multi_clients_group_removal_behavior(self):
        """Verify the p2p group removal behavior

        Steps:
        1. form a group between 3 peers
        2. verify their connection
        3. disconnect
        4. form a group between 3 peers
        5. trigger disconnect from GO
        6. check the group is removed
        7. form a group between 3 peers
        8. trigger disconnect from a GC
        9. check the group is still alive
        10. disconnect
        """
        all_duts = [self.dut1, self.dut2, self.dut3]

        # the 1st round
        (go_dut, gc_dut) = self.form_group(self.dut1, self.dut2)
        gc2_dut = self.dut3
        wp2putils.p2p_connect(gc2_dut,
                              go_dut,
                              False,
                              WPS_PBC,
                              p2p_connect_type=p2pconsts.P2P_CONNECT_JOIN)

        self.verify_group_connection([gc_dut, gc2_dut])

        go_dut.log.info("Trigger disconnection")
        wp2putils.p2p_disconnect(go_dut)
        self.check_disconnection([gc_dut, gc2_dut])
        time.sleep(p2pconsts.DEFAULT_FUNCTION_SWITCH_TIME)

        # the 2nd round
        self.log.info("Reconnect test, triggered by GO")
        self.clear_all_events(all_duts)

        self.form_group(go_dut, gc_dut, isReconnect=True)
        wp2putils.p2p_connect(gc2_dut,
                              go_dut,
                              True,
                              WPS_PBC,
                              p2p_connect_type=p2pconsts.P2P_CONNECT_JOIN)

        # trigger disconnect from GO, the group is destroyed and all
        # client are disconnected.
        go_dut.log.info("Trigger disconnection")
        wp2putils.p2p_disconnect(go_dut)
        self.check_disconnection([gc_dut, gc2_dut])
        time.sleep(p2pconsts.DEFAULT_FUNCTION_SWITCH_TIME)

        # the 3rd round
        self.log.info("Reconnect test, triggered by GC")
        self.clear_all_events(all_duts)

        self.form_group(go_dut, gc_dut, isReconnect=True)
        wp2putils.p2p_connect(gc2_dut,
                              go_dut,
                              True,
                              WPS_PBC,
                              p2p_connect_type=p2pconsts.P2P_CONNECT_JOIN)

        # trigger disconnect from GC, the group is still there.
        gc_dut.log.info("Trigger disconnection")
        wp2putils.p2p_disconnect(gc_dut)
        self.verify_group_connection([
            gc2_dut,
        ])

        # all clients are disconnected, the group is removed.
        wp2putils.p2p_disconnect(gc2_dut)
        self.check_disconnection([
            go_dut,
        ])
        time.sleep(p2pconsts.DEFAULT_FUNCTION_SWITCH_TIME)

    @test_tracker_info(uuid="6ea6e802-df62-4ae2-aa15-44c3267fd99b")
    def test_p2p_connect_with_p2p_and_join_go(self):
        """Verify the invitation from GC

        Steps:
        1. form a group between 2 peers
        2. gc joins the group via go
        2. verify their connection
        3. disconnect
        """
        all_duts = [self.dut1, self.dut2, self.dut3]

        (go_dut, gc_dut) = self.form_group(self.dut1, self.dut2)
        gc2_dut = self.dut3
        wp2putils.p2p_connect(gc2_dut,
                              go_dut,
                              False,
                              WPS_PBC,
                              p2p_connect_type=p2pconsts.P2P_CONNECT_JOIN)

        self.verify_group_connection([gc_dut, gc2_dut])

        go_dut.log.info("Trigger disconnection")
        wp2putils.p2p_disconnect(go_dut)
        self.check_disconnection([gc_dut, gc2_dut])
        time.sleep(p2pconsts.DEFAULT_FUNCTION_SWITCH_TIME)

    @test_tracker_info(uuid="e00469a4-93b7-44dd-8a5e-5d317e0e9333")
    def test_p2p_connect_with_p2p_and_legacy_client(self):
        """Verify the invitation from GC

        Steps:
        1. form a group between 2 peers
        2. gc joins the group via go
        2. verify their connection
        3. disconnect
        """
        all_duts = [self.dut1, self.dut2, self.dut3]

        (go_dut, gc_dut) = self.form_group(self.dut1, self.dut2)
        gc2_dut = self.dut3

        group = wp2putils.p2p_get_current_group(go_dut)
        network = {
            WifiEnums.SSID_KEY: group['NetworkName'],
            WifiEnums.PWD_KEY: group['Passphrase']
        }
        wutils.start_wifi_connection_scan_and_ensure_network_found(
            gc2_dut, group['NetworkName'])
        wutils.wifi_connect(gc2_dut, network, num_of_tries=3)

        go_dut.log.info("Trigger disconnection")
        wp2putils.p2p_disconnect(go_dut)
        time.sleep(p2pconsts.DEFAULT_FUNCTION_SWITCH_TIME)

    @test_tracker_info(uuid="bd53cc18-bcc7-4e27-b78b-1506f5c098c5")
    def test_p2p_connect_with_p2p_and_go_invite_peer(self):
        """Verify the invitation from GC

        Steps:
        1. form a group between 2 peers
        2. gc joins the group via go
        2. verify their connection
        3. disconnect
        """
        all_duts = [self.dut1, self.dut2, self.dut3]

        (go_dut, gc_dut) = self.form_group(self.dut1, self.dut2)
        gc2_dut = self.dut3
        wp2putils.p2p_connect(
            go_dut,
            gc2_dut,
            False,
            WPS_PBC,
            p2p_connect_type=p2pconsts.P2P_CONNECT_INVITATION,
            go_ad=go_dut)

        self.verify_group_connection([gc_dut, gc2_dut])

        go_dut.log.info("Trigger disconnection")
        wp2putils.p2p_disconnect(go_dut)
        self.check_disconnection([gc_dut, gc2_dut])
        time.sleep(p2pconsts.DEFAULT_FUNCTION_SWITCH_TIME)

    @test_tracker_info(uuid="4d6e666d-dc48-4881-86c1-5d7cec5e2571")
    def test_p2p_connect_with_p2p_and_gc_invite_peer(self):
        """Verify the invitation from GC

        Steps:
        1. form a group between 2 peers
        2. gc joins the group via go
        2. verify their connection
        3. disconnect
        """
        all_duts = [self.dut1, self.dut2, self.dut3]

        (go_dut, gc_dut) = self.form_group(self.dut1, self.dut2)
        gc2_dut = self.dut3
        wp2putils.p2p_connect(
            gc_dut,
            gc2_dut,
            False,
            WPS_PBC,
            p2p_connect_type=p2pconsts.P2P_CONNECT_INVITATION,
            go_ad=go_dut)

        self.verify_group_connection([gc_dut, gc2_dut])

        go_dut.log.info("Trigger disconnection")
        wp2putils.p2p_disconnect(go_dut)
        self.check_disconnection([gc_dut, gc2_dut])
        time.sleep(p2pconsts.DEFAULT_FUNCTION_SWITCH_TIME)