• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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_contrib.test_utils.wifi.wifi_test_utils as wutils
18import acts.utils
19import time
20import acts.controllers.packet_capture as packet_capture
21import re
22import logging
23#import acts.signals as signals
24from acts import asserts
25from acts import utils
26
27from acts.test_decorators import test_tracker_info
28from acts_contrib.test_utils.wifi.p2p.WifiP2pBaseTest import WifiP2pBaseTest
29from acts_contrib.test_utils.wifi.p2p import wifi_p2p_test_utils as wp2putils
30from acts_contrib.test_utils.wifi.p2p import wifi_p2p_const as p2pconsts
31from acts.controllers.ap_lib.hostapd_constants import CHANNEL_MAP
32
33
34WPS_PBC = wp2putils.WifiP2PEnums.WpsInfo.WIFI_WPS_INFO_PBC
35WPS_DISPLAY = wp2putils.WifiP2PEnums.WpsInfo.WIFI_WPS_INFO_DISPLAY
36WPS_KEYPAD = wp2putils.WifiP2PEnums.WpsInfo.WIFI_WPS_INFO_KEYPAD
37
38WifiEnums = wutils.WifiEnums
39
40class WifiP2pMultiCoutryTest(WifiP2pBaseTest):
41    """Tests for APIs in Android's WifiP2pManager class.
42
43    Test Bed Requirement:
44    * At least two Android devices
45    * 3 Android devices for WifiP2pMultiPeersTest.py
46    """
47    def __init__(self, controllers):
48        WifiP2pBaseTest.__init__(self, controllers)
49        self.basetest_name = (
50            "test_p2p_connect_via_display_and_ping_and_reconnect_multicountry",
51            "test_p2p_connect_via_pbc_and_ping_and_reconnect_multicountry",
52            )
53
54        self.generate_tests()
55
56    def generate_testcase(self, basetest_name, country):
57        """Generates a single test case from the given data.
58
59        Args:
60            basetest_name: The name of the base test case.
61            country: The information about the country code under test.
62        """
63        base_test = getattr(self, basetest_name)
64        test_tracker_uuid = ""
65
66        testcase_name = 'test_%s_%s' % (basetest_name, country)
67        test_case = test_tracker_info(uuid=test_tracker_uuid)(
68            lambda: base_test(country))
69        setattr(self, testcase_name, test_case)
70        self.tests.append(testcase_name)
71
72    def generate_tests(self):
73        for country in self.user_params['wifi_country_code']:
74                for basetest_name in self.basetest_name:
75                    self.generate_testcase(basetest_name, country)
76
77    def setup_class(self):
78        super().setup_class()
79        if hasattr(self, 'packet_capture'):
80            self.packet_capture = self.packet_capture[0]
81        self.channel_list_2g = WifiEnums.ALL_2G_FREQUENCIES
82        self.channel_list_5g = WifiEnums.ALL_5G_FREQUENCIES
83
84    def conf_packet_capture(self, band, channel):
85        """Configure packet capture on necessary channels."""
86        freq_to_chan = wutils.WifiEnums.freq_to_channel[int(channel)]
87        logging.info("Capturing packets from "
88                     "frequency:{}, Channel:{}".format(channel, freq_to_chan))
89        result = self.packet_capture.configure_monitor_mode(band, freq_to_chan)
90        if not result:
91            logging.error("Failed to configure channel "
92                          "for {} band".format(band))
93        self.pcap_procs = wutils.start_pcap(
94            self.packet_capture, band, self.test_name)
95        time.sleep(5)
96
97    def get_p2p0_freq(self, dut):
98        """ get P2P0 interface status"""
99        get_p2p0 = "timeout 3 logcat |grep P2P-GROUP-STARTED |grep p2p-p2p0-* |grep freq="
100        out_p2p01 = dut.adb.shell(get_p2p0)
101        out_p2p0 = re.findall("freq=(\d+)", out_p2p01)
102        return out_p2p0
103
104    """Test Cases"""
105
106    @test_tracker_info(uuid="f7d98b9e-494e-4e60-ae29-8418e270d2d8")
107    def test_p2p_connect_via_pbc_and_ping_and_reconnect_multicountry(self, country):
108        """Verify the p2p connect via pbc functionality
109
110        Steps:
111        1. Setting country code for each device
112        2. Request the connection which include discover the target device
113        3. check which dut is GO and which dut is GC
114        4. connection check via ping from GC to GO
115        5. disconnect
116        6. Trigger connect again from GO for reconnect test.
117        7. GO trigger disconnect
118        8. Trigger connect again from GC for reconnect test.
119        9. GC trigger disconnect
120        """
121        # Request the connection
122
123        wutils.set_wifi_country_code(self.dut1, country)
124        wutils.set_wifi_country_code(self.dut2, country)
125        wp2putils.p2p_connect(self.dut1, self.dut2, False, WPS_PBC)
126        p2pfreg = int(self.get_p2p0_freq(self.dut1)[0])
127        p2p_channel = str(CHANNEL_MAP[p2pfreg])
128        n = int(p2p_channel)
129        if n in range(len(self.channel_list_2g)):
130            sp2p_band = '2g'
131        else:
132            sp2p_band = '5g'
133        self.dut1.log.info('p2p frequency : {}'.format(p2pfreg))
134        self.dut1.log.info('p2p channel : {}'.format(p2p_channel))
135        self.dut1.log.info('p2p band : {}'.format(sp2p_band))
136        if hasattr(self, 'packet_capture'):
137            self.conf_packet_capture(sp2p_band, p2pfreg)
138        if wp2putils.is_go(self.dut1):
139            go_dut = self.dut1
140            gc_dut = self.dut2
141        elif wp2putils.is_go(self.dut2):
142            go_dut = self.dut2
143            gc_dut = self.dut1
144
145        go_ip = wp2putils.p2p_go_ip(gc_dut)
146        wp2putils.p2p_connection_ping_test(gc_dut, go_ip)
147
148        # trigger disconnect
149        wp2putils.p2p_disconnect(self.dut1)
150        wp2putils.check_disconnect(self.dut2)
151        time.sleep(p2pconsts.DEFAULT_FUNCTION_SWITCH_TIME)
152
153        self.log.info("Reconnect test, triggered by GO")
154        # trigger reconnect from GO
155        go_dut.ed.clear_all_events()
156        gc_dut.ed.clear_all_events()
157        wp2putils.p2p_connect(go_dut, gc_dut, True, WPS_PBC)
158        wp2putils.p2p_disconnect(go_dut)
159        wp2putils.check_disconnect(gc_dut)
160        time.sleep(p2pconsts.DEFAULT_FUNCTION_SWITCH_TIME)
161
162        # trigger reconnect from GC
163        self.log.info("Reconnect test, triggered by GC")
164        go_dut.ed.clear_all_events()
165        gc_dut.ed.clear_all_events()
166        wp2putils.p2p_connect(gc_dut, go_dut, True, WPS_PBC)
167        wp2putils.p2p_disconnect(gc_dut)
168        wp2putils.check_disconnect(
169            go_dut, timeout=p2pconsts.DEFAULT_GROUP_CLIENT_LOST_TIME)
170        time.sleep(p2pconsts.DEFAULT_FUNCTION_SWITCH_TIME)
171
172        if hasattr(self, 'packet_capture'):
173            wutils.stop_pcap(self.packet_capture, self.pcap_procs, False)
174        time.sleep(10)
175
176    @test_tracker_info(uuid="56b9745a-06ec-49fc-91d5-383f2dac082a")
177    def test_p2p_connect_via_display_and_ping_and_reconnect_multicountry(self, country):
178        """Verify the p2p connect via display functionality
179
180        Steps:
181        1. Setting country code for each device
182        2. Request the connection which include discover the target device
183        3. check which dut is GO and which dut is GC
184        4. connection check via ping from GC to GO
185        5. disconnect
186        6. Trigger connect again from GO for reconnect test.
187        7. GO trigger disconnect
188        8. Trigger connect again from GC for reconnect test.
189        9. GC trigger disconnect
190        """
191        # Request the connection
192        wutils.set_wifi_country_code(self.dut1, country)
193        wutils.set_wifi_country_code(self.dut2, country)
194        wp2putils.p2p_connect(self.dut1, self.dut2, False, WPS_DISPLAY)
195
196        p2pfreg = int(self.get_p2p0_freq(self.dut1)[0])
197        p2p_channel = str(CHANNEL_MAP[p2pfreg])
198        n = int(p2p_channel)
199        if n in range(len(self.channel_list_2g)):
200            sp2p_band = '2g'
201        else:
202            sp2p_band = '5g'
203        self.dut1.log.info('p2p frequency : {}'.format(p2pfreg))
204        self.dut1.log.info('p2p channel : {}'.format(p2p_channel))
205        self.dut1.log.info('p2p band : {}'.format(sp2p_band))
206        if hasattr(self, 'packet_capture'):
207            self.conf_packet_capture(sp2p_band, p2pfreg)
208        if wp2putils.is_go(self.dut1):
209            go_dut = self.dut1
210            gc_dut = self.dut2
211        elif wp2putils.is_go(self.dut2):
212            go_dut = self.dut2
213            gc_dut = self.dut1
214
215        go_ip = wp2putils.p2p_go_ip(gc_dut)
216        wp2putils.p2p_connection_ping_test(gc_dut, go_ip)
217
218        # trigger disconnect
219        wp2putils.p2p_disconnect(self.dut1)
220        wp2putils.check_disconnect(self.dut2)
221        time.sleep(p2pconsts.DEFAULT_FUNCTION_SWITCH_TIME)
222
223        self.log.info("Reconnect test, triggered by GO")
224        # trigger reconnect from GO
225        go_dut.ed.clear_all_events()
226        gc_dut.ed.clear_all_events()
227        wp2putils.p2p_connect(go_dut, gc_dut, True, WPS_DISPLAY)
228        wp2putils.p2p_disconnect(go_dut)
229        wp2putils.check_disconnect(gc_dut)
230        time.sleep(p2pconsts.DEFAULT_FUNCTION_SWITCH_TIME)
231
232        # trigger reconnect from GC
233        self.log.info("Reconnect test, triggered by GC")
234        go_dut.ed.clear_all_events()
235        gc_dut.ed.clear_all_events()
236        wp2putils.p2p_connect(gc_dut, go_dut, True, WPS_DISPLAY)
237        wp2putils.p2p_disconnect(gc_dut)
238        wp2putils.check_disconnect(
239            go_dut, timeout=p2pconsts.DEFAULT_GROUP_CLIENT_LOST_TIME)
240        time.sleep(p2pconsts.DEFAULT_FUNCTION_SWITCH_TIME)
241        if hasattr(self, 'packet_capture'):
242            wutils.stop_pcap(self.packet_capture, self.pcap_procs, False)
243        time.sleep(10)
244