• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#  Copyright (C) 2025 The Android Open Source Project
2#
3#  Licensed under the Apache License, Version 2.0 (the "License");
4#  you may not use this file except in compliance with the License.
5#  You may obtain a copy of the License at
6#
7#       http://www.apache.org/licenses/LICENSE-2.0
8#
9#  Unless required by applicable law or agreed to in writing, software
10#  distributed under the License is distributed on an "AS IS" BASIS,
11#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12#  See the License for the specific language governing permissions and
13#  limitations under the License.
14#
15#  Licensed under the Apache License, Version 2.0 (the "License");
16#  you may not use this file except in compliance with the License.
17#  You may obtain a copy of the License at
18#
19#       http://www.apache.org/licenses/LICENSE-2.0
20#
21#  Unless required by applicable law or agreed to in writing, software
22#  distributed under the License is distributed on an "AS IS" BASIS,
23#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
24#  See the License for the specific language governing permissions and
25#  limitations under the License.
26
27# Lint as: python3
28"""ACTS Wifi P2pManager Test reimplemented in Mobly."""
29
30from collections.abc import Sequence
31import datetime
32import logging
33import time
34
35from android.platform.test.annotations import ApiTest
36from direct import constants
37from direct import p2p_utils
38from mobly import base_test
39from mobly import records
40from mobly import test_runner
41from mobly import utils
42from mobly.controllers import android_device
43import wifi_p2p_lib as wp2putils
44
45
46_DEFAULT_TIMEOUT = datetime.timedelta(seconds=30)
47DEFAULT_SLEEPTIME = 5
48_DEFAULT_FUNCTION_SWITCH_TIME = 10
49_DEFAULT_GROUP_CLIENT_LOST_TIME = 60
50
51_WIFI_DIRECT_SNIPPET_KEY = 'wifi_direct_mobly_snippet'
52
53P2P_CONNECT_NEGOTIATION = 0
54P2P_CONNECT_JOIN = 1
55P2P_CONNECT_INVITATION = 2
56
57WPS_PBC = wp2putils.WifiP2PEnums.WpsInfo.WIFI_WPS_INFO_PBC
58WPS_DISPLAY = wp2putils.WifiP2PEnums.WpsInfo.WIFI_WPS_INFO_DISPLAY
59WPS_KEYPAD = wp2putils.WifiP2PEnums.WpsInfo.WIFI_WPS_INFO_KEYPAD
60
61
62class WifiP2pManagerTest(base_test.BaseTestClass):
63    """Tests Wi-Fi Direct between 2 Android devices."""
64
65    ads: Sequence[android_device.AndroidDevice]
66    group_owner_ad: android_device.AndroidDevice
67    client_ad: android_device.AndroidDevice
68    network_name = 'DIRECT-xy-Hello'
69    passphrase = 'P2pWorld1234'
70    group_band = '2'
71
72    def setup_class(self) -> None:
73        super().setup_class()
74        self.ads = self.register_controller(android_device, min_number=2)
75        utils.concurrent_exec(
76            self._setup_device,
77            param_list=[[ad] for ad in self.ads],
78            raise_on_exception=True,
79        )
80        self.group_owner_ad, self.client_ad, *_ = self.ads
81        self.group_owner_ad.debug_tag = (
82            f'{self.group_owner_ad.serial}(Group Owner)'
83        )
84        self.client_ad.debug_tag = f'{self.client_ad.serial}(Client)'
85
86    def _setup_device(self, ad: android_device.AndroidDevice) -> None:
87        ad.load_snippet('wifi', constants.WIFI_SNIPPET_PACKAGE_NAME)
88        # Clear all saved Wi-Fi networks.
89        ad.wifi.wifiDisable()
90        ad.wifi.wifiClearConfiguredNetworks()
91        ad.wifi.wifiEnable()
92
93    def _teardown_wifi_p2p(self, ad: android_device.AndroidDevice):
94        try:
95            p2p_utils.teardown_wifi_p2p(ad)
96        finally:
97            ad.services.create_output_excerpts_all(self.current_test_info)
98
99    def teardown_test(self) -> None:
100        utils.concurrent_exec(
101            self._teardown_wifi_p2p,
102            param_list=[[ad] for ad in self.ads],
103            raise_on_exception=True,
104        )
105
106    def on_fail(self, record: records.TestResult) -> None:
107        logging.info('Collecting bugreports...')
108        android_device.take_bug_reports(
109            self.ads, destination=self.current_test_info.output_path
110        )
111
112    @ApiTest(
113        apis=[
114            'android.net.wifi.p2p.WifiP2pManager#discoverPeers(android.net.wifi.p2p.WifiP2pManager.Channel channel, android.net.wifi.p2p.WifiP2pManager.ActionListener listener)',
115        ]
116    )
117    def test_p2p_discovery(self):
118        """Verify the p2p discovery functionality.
119
120        Steps:
121        1. Discover the target device
122        2. Check the target device in peer list
123        """
124        self.ads[0].log.info('Device discovery')
125        responder = p2p_utils.setup_wifi_p2p(self.ads[0])
126        requester = p2p_utils.setup_wifi_p2p(self.ads[1])
127
128        requester.ad.log.info('Searching for target device.')
129        responder_p2p_dev = p2p_utils.discover_p2p_peer(responder, requester)
130        self.ads[0].log.info('name= %s, address=%s, group_owner=%s',
131                             responder_p2p_dev.device_name,
132                             responder_p2p_dev.device_address,
133                             responder_p2p_dev.is_group_owner)
134        requester_p2p_dev = p2p_utils.discover_p2p_peer(requester, responder)
135        self.ads[1].log.info('name= %s, address=%s, group_owner=%s',
136                             requester_p2p_dev.device_name,
137                             requester_p2p_dev.device_address,
138                             requester_p2p_dev.is_group_owner)
139
140    @ApiTest(
141        apis=[
142            'android.net.wifi.p2p.WifiP2pManager#requestConnectionInfo(android.net.wifi.p2p.WifiP2pManager.Channel channel, android.net.wifi.p2p.WifiP2pManager.ConnectionInfoListener listener)',
143            'android.net.wifi.p2p.WifiP2pManager#connect(android.net.wifi.p2p.WifiP2pManager.Channel channel, android.net.wifi.p2p.WifiP2pConfig config, android.net.wifi.p2p.WifiP2pManager.ActionListener listener)',
144        ]
145    )
146    def test_p2p_connect_via_pbc_and_ping_and_reconnect(self):
147        """Verify the p2p connect via pbc functionality.
148
149        Steps:
150        1. Request the connection which include discover the target device
151        2. check which dut is GO and which dut is GC
152        3. connection check via ping from GC to GO
153        4. disconnect
154        5. Trigger connect again from GO for reconnect test.
155        6. GO trigger disconnect
156        7. Trigger connect again from GC for reconnect test.
157        8. GC trigger disconnect
158        """
159        self.ads[0].log.info('Device initialize')
160        go_dut = self.ads[0]
161        gc_dut = self.ads[1]
162
163        logging.info('GO: %s, GC: %s', go_dut.serial, gc_dut.serial)
164        device_go = p2p_utils.setup_wifi_p2p(go_dut)
165        device_gc = p2p_utils.setup_wifi_p2p(gc_dut)
166        self.run_p2p_connect_and_ping(device_gc, device_go, WPS_PBC, False)
167        self.run_p2p_connect_and_ping(device_go, device_gc, WPS_PBC, True)
168        self.run_p2p_connect_and_ping(device_gc, device_go, WPS_PBC, True)
169
170    @ApiTest(
171        apis=[
172            'android.net.wifi.p2p.WifiP2pManager#requestConnectionInfo(android.net.wifi.p2p.WifiP2pManager.Channel channel, android.net.wifi.p2p.WifiP2pManager.ConnectionInfoListener listener)',
173            'android.net.wifi.p2p.WifiP2pManager#connect(android.net.wifi.p2p.WifiP2pManager.Channel channel, android.net.wifi.p2p.WifiP2pConfig config, android.net.wifi.p2p.WifiP2pManager.ActionListener listener)',
174        ]
175    )
176    def test_p2p_connect_via_display_and_ping_and_reconnect(self):
177        """Verify the p2p connect via display functionality.
178
179        Steps:
180        1. Request the connection which include discover the target device
181        2. check which dut is GO and which dut is GC
182        3. connection check via ping from GC to GO
183        4. disconnect
184        5. Trigger connect again from GO for reconnect test.
185        6. GO trigger disconnect
186        7. Trigger connect again from GC for reconnect test.
187        8. GC trigger disconnect
188        """
189        self.ads[0].log.info('Device initialize')
190        go_dut = self.ads[0]
191        gc_dut = self.ads[1]
192
193        logging.info('GO: %s, GC: %s', go_dut.serial, gc_dut.serial)
194        device_go = p2p_utils.setup_wifi_p2p(go_dut)
195        device_gc = p2p_utils.setup_wifi_p2p(gc_dut)
196        self.run_p2p_connect_and_ping(device_gc, device_go, WPS_DISPLAY, False)
197        self.run_p2p_connect_and_ping(device_go, device_gc, WPS_DISPLAY, True)
198        self.run_p2p_connect_and_ping(device_gc, device_go, WPS_DISPLAY, True)
199
200    @ApiTest(
201        apis=[
202            'android.net.wifi.p2p.WifiP2pManager#requestConnectionInfo(android.net.wifi.p2p.WifiP2pManager.Channel channel, android.net.wifi.p2p.WifiP2pManager.ConnectionInfoListener listener)',
203            'android.net.wifi.p2p.WifiP2pManager#connect(android.net.wifi.p2p.WifiP2pManager.Channel channel, android.net.wifi.p2p.WifiP2pConfig config, android.net.wifi.p2p.WifiP2pManager.ActionListener listener)',
204        ]
205    )
206    def test_p2p_connect_via_keypad_and_ping_and_reconnect(self):
207        """Verify the p2p connect via keypad functionality.
208
209        Steps:
210        1. Request the connection which include discover the target device
211        2. check which dut is GO and which dut is GC
212        3. connection check via ping from GC to GO
213        4. disconnect
214        5. Trigger connect again from GO for reconnect test.
215        6. GO trigger disconnect
216        7. Trigger connect again from GC for reconnect test.
217        8. GC trigger disconnect
218        """
219        self.ads[0].log.info('Device initialize')
220        go_dut = self.ads[0]
221        gc_dut = self.ads[1]
222
223        logging.info('GO: %s, GC: %s', go_dut.serial, gc_dut.serial)
224        device_go = p2p_utils.setup_wifi_p2p(go_dut)
225        device_gc = p2p_utils.setup_wifi_p2p(gc_dut)
226
227        self.run_p2p_connect_and_ping(device_gc, device_go, WPS_KEYPAD, False)
228        self.run_p2p_connect_and_ping(device_go, device_gc, WPS_KEYPAD, True)
229        self.run_p2p_connect_and_ping(device_gc, device_go, WPS_KEYPAD, True)
230
231    def run_p2p_connect_and_ping(
232        self,
233        device1,
234        device2,
235        pws_method,
236        re_connect):
237        # Request the connection
238        wp2putils.p2p_connect(device1, device2, re_connect, pws_method)
239
240        if wp2putils.is_go(device1.ad):
241            client_dut = device2.ad
242        else:
243            client_dut = device1.ad
244        logging.info('Client is : %s', client_dut.serial)
245        go_ip = wp2putils.p2p_go_ip(client_dut)
246        wp2putils.p2p_connection_ping_test(client_dut, go_ip)
247
248        # trigger disconnect
249        p2p_utils.remove_group_and_verify_disconnected(
250            device1, device2, is_group_negotiation=False
251        )
252        time.sleep(_DEFAULT_FUNCTION_SWITCH_TIME)
253
254if __name__ == '__main__':
255  test_runner.main()
256
257