• 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 csv
18import json
19import os
20import time
21import acts_contrib.test_utils.wifi.wifi_test_utils as wutils
22from acts import signals
23from acts.test_decorators import test_tracker_info
24from acts_contrib.test_utils.wifi.WifiBaseTest import WifiBaseTest
25from acts_contrib.test_utils.wifi import pdu_controller_utils
26
27WifiEnums = wutils.WifiEnums
28WAIT_BETWEEN_ACTIONS = 5
29
30class WifiIOTConnectionTest(WifiBaseTest):
31    """ Tests for wifi IOT Connection
32
33        Test Bed Requirement:
34          * One Android device
35          * Wi-Fi IOT networks visible to the device
36    """
37
38    def __init__(self, controllers):
39        WifiBaseTest.__init__(self, controllers)
40        self.generate_test_list()
41
42    def setup_class(self):
43        self.dut = self.android_devices[0]
44        self.pdu = pdu_controller_utils.create(self.user_params['Pdu'])[0]
45
46        if hasattr(self, 'packet_capture'):
47            self.packet_capture = self.packet_capture[0]
48
49        wutils.wifi_test_device_init(self.dut)
50
51        self.csv_write(["Project", "TimeStamp", "Build_ID", "Test_Name",
52                        "Host", "RSSI", "Link_speed", "freq", "#iteration",
53                        "#Pass"], )
54
55        req_params = ['iot_password', 'pdu_wait_time',
56                      'iot_ssid', 'Pdu', 'iot_connection_iteration']
57        self.unpack_userparams(
58            req_param_names=req_params)
59
60    def setup_test(self):
61        self.dut.droid.wakeLockAcquireBright()
62        self.dut.droid.wakeUpNow()
63        wutils.reset_wifi(self.dut)
64
65    def teardown_test(self):
66        self.dut.droid.wakeLockRelease()
67        self.dut.droid.goToSleepNow()
68        wutils.stop_pcap(self.packet_capture, self.pcap_procs, False)
69
70    def on_fail(self, test_name, begin_time):
71        self.dut.take_bug_report(test_name, begin_time)
72        self.dut.cat_adb_log(test_name, begin_time)
73
74    def csv_write(self, data):
75        """Output .CSV file as a result.
76
77        Args:
78            data: a dict containing:
79                'project', 'TimeStamp', 'test_name', 'test_name',
80                'host', 'rssi', 'link_speed', 'freq', '#Iteration', '#Pass'
81        """
82        time_str = time.strftime("%Y-%m-%d")
83        file_name_format = "wifi_iot_connection_" + time_str + ".csv"
84        file_name = os.path.join(self.log_path, file_name_format)
85        with open(file_name, "a", newline="") as csv_file:
86            csv_writer = csv.writer(csv_file, delimiter=',')
87            csv_writer.writerow(data)
88
89    def get_wifi_info(self):
90        """Get current connected WiFi AP's info.
91
92        Returns:
93            rssi: Wi-Fi rssi
94            link_speed: Wi-Fi link speed
95            freq: Wi-Fi freq
96        """
97        freq = "NA"
98        link_speed = "NA"
99        rssi = "NA"
100
101        try:
102            out = self.dut.adb.shell("iw wlan0 link")
103            if out:
104                for line in out.split("\n"):
105                    if "freq:" in line:
106                        freq = line.split()[1]
107                    elif "signal" in line:
108                        rssi = line.split()[1]
109                    elif "bitrate" in line:
110                        link_speed = line.split()[2]
111        except AttributeError as e:
112            self.log.debug("No wifi info, check Wi-Fi connection.", e)
113        finally:
114            return rssi, link_speed, freq
115
116    def getprop(self, property_name):
117        """Get dut's property
118
119        Args:
120            property_name: property name, e.g., "ro.build.product"
121        Return:
122            property
123        """
124        return self.dut.adb.getprop(property_name)
125
126    def generate_test_list(self):
127        """Generates a test list which is sorted by host number."""
128        sorted_list = sorted(
129            self.user_params["iot_ssid"], key=lambda ssid: ssid["host"])
130        for test_item in sorted_list:
131            self.init_test_case(self.wifi_iot_connection_test, test_item)
132
133    def init_test_case(self, wifi_iot_connection_test, test_item):
134        """Generates a single test case from the given data.
135
136        Args:
137            wifi_iot_connection_test: The base test case function to run.
138            test_item: test case required info, include:
139                "ssid", "uuid", "host", "band", "channel"
140        """
141        test_name = test_item["ssid"]
142        test_tracker_uuid = test_item["uuid"]
143        if not test_name.startswith("test_"):
144            test_name = "test_{}".format(test_name)
145        test_case = test_tracker_info(uuid=test_tracker_uuid)(
146            lambda: wifi_iot_connection_test(test_item,
147                                             self.user_params["iot_password"]))
148        setattr(self, test_name, test_case)
149        self.tests.append(test_name)
150
151    def pdu_status_check(self):
152        """ Check pdu status
153        if pdu's currently ON outlet != next test case's host number,
154        switch ON outlet to the host number.
155        for example: pdu's outlet 3 is on, next test case host number is 4,
156        turn off pdu outlet 3 and turn on outlet 4.
157        """
158        _, self.pdu_status = self.pdu.get_status()
159        self.log.info("Current host of the pdu : {}".format(self.pdu_status))
160        self.log.info("Next host of the test case : {}".format(self.host))
161        if str(self.host) != self.pdu_status:
162            self.pdu.off_all()
163            self.log.info("Switch PDU to {}".format(self.host))
164            self.pdu.turn_on_outlets(str(self.host))
165            self.log.info("Wait {} secs to bring up the APs"
166                          .format(self.user_params["pdu_wait_time"]))
167            time.sleep(self.user_params["pdu_wait_time"])
168
169    def start_packet_capture(self, band, channel):
170        """Configure wireless packet capture on pre-defined channels.
171
172        Args:
173            band: '2G' or '5G'. pre-defined in config file.
174            channel: Wi-fi Channel, pre-defined in the config file.
175        """
176        self.log.info("Capturing packets from Channel: {}".format(channel))
177        result = self.packet_capture.configure_monitor_mode(band, channel)
178        if not result:
179            self.dut.log.error("Failed to configure channel "
180                               "for {} band".format(band))
181        self.pcap_procs = wutils.start_pcap(
182            self.packet_capture, band, self.current_test_name)
183        time.sleep(3)
184
185    def ping_public_gateway_ip(self):
186        """Ping 8.8.8.8"""
187        try:
188            ping_result = self.dut.adb.shell('ping -w 5 8.8.8.8')
189            if '0%' in ping_result:
190                self.dut.log.info('Ping success')
191            return True
192        except:
193            self.dut.log.error('Faild to ping public gateway 8.8.8.8')
194            return False
195
196    def connect_to_network_and_ping(self, network):
197        ssid = network[WifiEnums.SSID_KEY]
198        connection_pass = 0
199        for i in range(self.user_params['iot_connection_iteration']):
200            self.dut.log.info('Connection iteration : {}'.format(i + 1))
201            try:
202                wutils.connect_to_wifi_network(self.dut, network,
203                        num_of_connect_tries=1,
204                        check_connectivity=False)
205                time.sleep(WAIT_BETWEEN_ACTIONS)
206                self.rssi, self.link_speed, self.freq = self.get_wifi_info()
207                time.sleep(WAIT_BETWEEN_ACTIONS)
208                if self.ping_public_gateway_ip():
209                    connection_pass += 1
210                wutils.wifi_forget_network(self.dut, ssid)
211                self.dut.log.info("connection_pass: {}"
212                                    .format(connection_pass))
213                time.sleep(WAIT_BETWEEN_ACTIONS)
214            except:
215                self.dut.log.error("connection_fail")
216
217        # Create a dictionary to store data in a json file.
218        connection_result = {}
219        connection_result["project"] = self.getprop("ro.build.product")
220        connection_result["TimeStamp"] = time.strftime("%Y-%m-%d %H:%M:%S",
221                                                       time.localtime())
222        connection_result["Build ID"] = self.getprop("ro.build.id")
223        connection_result["test_name"] = self.current_test_name
224        connection_result["host"] = self.host
225        connection_result['rssi'] = self.rssi
226        connection_result['link_speed'] = self.link_speed
227        connection_result['freq'] = self.freq
228        connection_result['#Iteration'] = self.user_params[
229            'iot_connection_iteration']
230        connection_result['#Pass'] = connection_pass
231
232        # Create a json file for each test case.
233        results_file_path = os.path.join(self.log_path, "{}.json".format(
234            self.current_test_name))
235
236        with open(results_file_path, 'w') as results_file:
237            json.dump(connection_result, results_file, indent=4)
238
239        data = (self.getprop("ro.build.product"),
240                time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
241                self.getprop("ro.build.id"),
242                connection_result["test_name"],
243                connection_result["host"],
244                connection_result['rssi'],
245                connection_result['link_speed'],
246                connection_result['freq'],
247                connection_result['#Iteration'],
248                connection_result['#Pass']
249                )
250
251        self.csv_write(data)
252
253        self.rssi = "NA"
254        self.link_speed = "NA"
255        self.freq = "NA"
256
257        if connection_pass < self.user_params['iot_connection_iteration']:
258            raise signals.TestFailure("connection failed more than expected")
259
260    def wifi_iot_connection_test(self, test_item, password):
261        """The base test case logic for Wifi IOT generated test cases.
262
263            1. Check pdu outlets is ON as expected.
264            2. Scan SSIDs of 10 Wi-Fi APs.
265            3. Start Packet Capture on pre-defined channel.
266            3. Connect to the AP and ping Gateway 8.8.8.8 for 5 times.
267
268        Args:
269            test_item: Test info include:
270                'ssid', 'host', 'uuid', 'band', 'channel'.
271            password: pwd for login to the access point.
272        """
273        network = {'SSID': test_item["ssid"], 'password': password}
274        self.host = test_item["host"]
275        self.pdu_status_check()
276
277        # Capturing wireless packets before DUT connect to Wi-Fi network.
278        self.band = test_item["band"]
279        self.channel = test_item["channel"]
280        if hasattr(self, 'packet_capture'):
281            self.start_packet_capture(self.band, self.channel)
282
283        # Connect to Wi-Fi network and ping public gateway for 5 times.
284        self.connect_to_network_and_ping(network)
285