1#!/usr/bin/env python3
2#
3#   Copyright 2020 - 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
17import acts.test_utils.wifi.wifi_test_utils as wutils
18import acts.utils
19import time
20
21from acts import asserts
22from acts import utils
23
24from acts.test_decorators import test_tracker_info
25from acts.test_utils.wifi import wifi_test_utils as wutils
26from acts.test_utils.wifi.p2p.WifiP2pBaseTest import WifiP2pBaseTest
27from acts.test_utils.wifi.p2p import wifi_p2p_test_utils as wp2putils
28from acts.test_utils.wifi.p2p import wifi_p2p_const as p2pconsts
29
30WPS_PBC = wp2putils.WifiP2PEnums.WpsInfo.WIFI_WPS_INFO_PBC
31WPS_DISPLAY = wp2putils.WifiP2PEnums.WpsInfo.WIFI_WPS_INFO_DISPLAY
32WPS_KEYPAD = wp2putils.WifiP2PEnums.WpsInfo.WIFI_WPS_INFO_KEYPAD
33
34WifiEnums = wutils.WifiEnums
35
36
37class WifiP2pMultiPeersTest(WifiP2pBaseTest):
38    """Tests for multiple clients.
39
40    Test Bed Requirement:
41    * 3 Android devices for each test in this class.
42    """
43    def __init__(self, controllers):
44        WifiP2pBaseTest.__init__(self, controllers)
45
46    def setup_test(self):
47        WifiP2pBaseTest.setup_test(self)
48        asserts.skip_if(
49            len(self.android_devices) < 3,
50            "No enough android devices. Skip this test")
51
52    def form_group(self, dut1, dut2, isReconnect=False, wpsType=WPS_PBC):
53        # Request the connection to create a group
54        wp2putils.p2p_connect(dut1, dut2, isReconnect, wpsType)
55
56        if wp2putils.is_go(dut1):
57            go_dut = dut1
58            gc_dut = dut2
59        elif wp2putils.is_go(dut2):
60            go_dut = dut2
61            gc_dut = dut1
62        return (go_dut, gc_dut)
63
64    def verify_group_connection(self, group_clients):
65        for gc in group_clients:
66            go_ip = wp2putils.p2p_go_ip(gc)
67            wp2putils.p2p_connection_ping_test(gc, go_ip)
68
69    def clear_all_events(self, duts):
70        for dut in duts:
71            dut.ed.clear_all_events()
72
73    def check_disconnection(self, duts):
74        for dut in duts:
75            wp2putils.check_disconnect(dut)
76
77    """Test Cases"""
78    @test_tracker_info(uuid="20cd4f4d-fe7d-4ee2-a832-33caa5b9700b")
79    def test_p2p_multi_clients_group_removal_behavior(self):
80        """Verify the p2p group removal behavior
81
82        Steps:
83        1. form a group between 3 peers
84        2. verify their connection
85        3. disconnect
86        4. form a group between 3 peers
87        5. trigger disconnect from GO
88        6. check the group is removed
89        7. form a group between 3 peers
90        8. trigger disconnect from a GC
91        9. check the group is still alive
92        10. disconnect
93        """
94        all_duts = [self.dut1, self.dut2, self.dut3]
95
96        # the 1st round
97        (go_dut, gc_dut) = self.form_group(self.dut1, self.dut2)
98        gc2_dut = self.dut3
99        wp2putils.p2p_connect(gc2_dut,
100                              go_dut,
101                              False,
102                              WPS_PBC,
103                              p2p_connect_type=p2pconsts.P2P_CONNECT_JOIN)
104
105        self.verify_group_connection([gc_dut, gc2_dut])
106
107        go_dut.log.info("Trigger disconnection")
108        wp2putils.p2p_disconnect(go_dut)
109        self.check_disconnection([gc_dut, gc2_dut])
110        time.sleep(p2pconsts.DEFAULT_FUNCTION_SWITCH_TIME)
111
112        # the 2nd round
113        self.log.info("Reconnect test, triggered by GO")
114        self.clear_all_events(all_duts)
115
116        self.form_group(go_dut, gc_dut, isReconnect=True)
117        wp2putils.p2p_connect(gc2_dut,
118                              go_dut,
119                              True,
120                              WPS_PBC,
121                              p2p_connect_type=p2pconsts.P2P_CONNECT_JOIN)
122
123        # trigger disconnect from GO, the group is destroyed and all
124        # client are disconnected.
125        go_dut.log.info("Trigger disconnection")
126        wp2putils.p2p_disconnect(go_dut)
127        self.check_disconnection([gc_dut, gc2_dut])
128        time.sleep(p2pconsts.DEFAULT_FUNCTION_SWITCH_TIME)
129
130        # the 3rd round
131        self.log.info("Reconnect test, triggered by GC")
132        self.clear_all_events(all_duts)
133
134        self.form_group(go_dut, gc_dut, isReconnect=True)
135        wp2putils.p2p_connect(gc2_dut,
136                              go_dut,
137                              True,
138                              WPS_PBC,
139                              p2p_connect_type=p2pconsts.P2P_CONNECT_JOIN)
140
141        # trigger disconnect from GC, the group is still there.
142        gc_dut.log.info("Trigger disconnection")
143        wp2putils.p2p_disconnect(gc_dut)
144        self.verify_group_connection([
145            gc2_dut,
146        ])
147
148        # all clients are disconnected, the group is removed.
149        wp2putils.p2p_disconnect(gc2_dut)
150        self.check_disconnection([
151            go_dut,
152        ])
153        time.sleep(p2pconsts.DEFAULT_FUNCTION_SWITCH_TIME)
154
155    @test_tracker_info(uuid="6ea6e802-df62-4ae2-aa15-44c3267fd99b")
156    def test_p2p_connect_with_p2p_and_join_go(self):
157        """Verify the invitation from GC
158
159        Steps:
160        1. form a group between 2 peers
161        2. gc joins the group via go
162        2. verify their connection
163        3. disconnect
164        """
165        all_duts = [self.dut1, self.dut2, self.dut3]
166
167        (go_dut, gc_dut) = self.form_group(self.dut1, self.dut2)
168        gc2_dut = self.dut3
169        wp2putils.p2p_connect(gc2_dut,
170                              go_dut,
171                              False,
172                              WPS_PBC,
173                              p2p_connect_type=p2pconsts.P2P_CONNECT_JOIN)
174
175        self.verify_group_connection([gc_dut, gc2_dut])
176
177        go_dut.log.info("Trigger disconnection")
178        wp2putils.p2p_disconnect(go_dut)
179        self.check_disconnection([gc_dut, gc2_dut])
180        time.sleep(p2pconsts.DEFAULT_FUNCTION_SWITCH_TIME)
181
182    @test_tracker_info(uuid="e00469a4-93b7-44dd-8a5e-5d317e0e9333")
183    def test_p2p_connect_with_p2p_and_legacy_client(self):
184        """Verify the invitation from GC
185
186        Steps:
187        1. form a group between 2 peers
188        2. gc joins the group via go
189        2. verify their connection
190        3. disconnect
191        """
192        all_duts = [self.dut1, self.dut2, self.dut3]
193
194        (go_dut, gc_dut) = self.form_group(self.dut1, self.dut2)
195        gc2_dut = self.dut3
196
197        group = wp2putils.p2p_get_current_group(go_dut)
198        network = {
199            WifiEnums.SSID_KEY: group['NetworkName'],
200            WifiEnums.PWD_KEY: group['Passphrase']
201        }
202        wutils.start_wifi_connection_scan_and_ensure_network_found(
203            gc2_dut, group['NetworkName'])
204        wutils.wifi_connect(gc2_dut, network, num_of_tries=3)
205
206        go_dut.log.info("Trigger disconnection")
207        wp2putils.p2p_disconnect(go_dut)
208        time.sleep(p2pconsts.DEFAULT_FUNCTION_SWITCH_TIME)
209
210    @test_tracker_info(uuid="bd53cc18-bcc7-4e27-b78b-1506f5c098c5")
211    def test_p2p_connect_with_p2p_and_go_invite_peer(self):
212        """Verify the invitation from GC
213
214        Steps:
215        1. form a group between 2 peers
216        2. gc joins the group via go
217        2. verify their connection
218        3. disconnect
219        """
220        all_duts = [self.dut1, self.dut2, self.dut3]
221
222        (go_dut, gc_dut) = self.form_group(self.dut1, self.dut2)
223        gc2_dut = self.dut3
224        wp2putils.p2p_connect(
225            go_dut,
226            gc2_dut,
227            False,
228            WPS_PBC,
229            p2p_connect_type=p2pconsts.P2P_CONNECT_INVITATION,
230            go_ad=go_dut)
231
232        self.verify_group_connection([gc_dut, gc2_dut])
233
234        go_dut.log.info("Trigger disconnection")
235        wp2putils.p2p_disconnect(go_dut)
236        self.check_disconnection([gc_dut, gc2_dut])
237        time.sleep(p2pconsts.DEFAULT_FUNCTION_SWITCH_TIME)
238
239    @test_tracker_info(uuid="4d6e666d-dc48-4881-86c1-5d7cec5e2571")
240    def test_p2p_connect_with_p2p_and_gc_invite_peer(self):
241        """Verify the invitation from GC
242
243        Steps:
244        1. form a group between 2 peers
245        2. gc joins the group via go
246        2. verify their connection
247        3. disconnect
248        """
249        all_duts = [self.dut1, self.dut2, self.dut3]
250
251        (go_dut, gc_dut) = self.form_group(self.dut1, self.dut2)
252        gc2_dut = self.dut3
253        wp2putils.p2p_connect(
254            gc_dut,
255            gc2_dut,
256            False,
257            WPS_PBC,
258            p2p_connect_type=p2pconsts.P2P_CONNECT_INVITATION,
259            go_ad=go_dut)
260
261        self.verify_group_connection([gc_dut, gc2_dut])
262
263        go_dut.log.info("Trigger disconnection")
264        wp2putils.p2p_disconnect(go_dut)
265        self.check_disconnection([gc_dut, gc2_dut])
266        time.sleep(p2pconsts.DEFAULT_FUNCTION_SWITCH_TIME)
267