1#!/usr/bin/env python3
2#
3#   Copyright 2019 - 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 inspect
18import logging
19
20import acts.test_utils.wifi.wifi_test_utils as awutils
21import acts.test_utils.abstract_devices.utils_lib.wlan_utils as fwutils
22from acts.utils import get_interface_ip_addresses
23from acts.utils import adb_shell_ping
24
25from acts import asserts
26from acts.controllers.fuchsia_device import FuchsiaDevice
27from acts.controllers.android_device import AndroidDevice
28
29
30def create_wlan_device(hardware_device):
31    """Creates a generic WLAN device based on type of device that is sent to
32    the functions.
33
34    Args:
35        hardware_device: A WLAN hardware device that is supported by ACTS.
36    """
37    if isinstance(hardware_device, FuchsiaDevice):
38        return FuchsiaWlanDevice(hardware_device)
39    elif isinstance(hardware_device, AndroidDevice):
40        return AndroidWlanDevice(hardware_device)
41    else:
42        raise ValueError('Unable to create WlanDevice for type %s' %
43                         type(hardware_device))
44
45
46FUCHSIA_VALID_SECURITY_TYPES = {"none", "wep", "wpa", "wpa2", "wpa3"}
47
48
49class WlanDevice(object):
50    """Class representing a generic WLAN device.
51
52    Each object of this class represents a generic WLAN device.
53    Android device and Fuchsia devices are the currently supported devices/
54
55    Attributes:
56        device: A generic WLAN device.
57    """
58    def __init__(self, device):
59        self.device = device
60        self.log = logging
61
62    def wifi_toggle_state(self, state):
63        """Base generic WLAN interface.  Only called if not overridden by
64        another supported device.
65        """
66        raise NotImplementedError("{} must be defined.".format(
67            inspect.currentframe().f_code.co_name))
68
69    def reset_wifi(self):
70        """Base generic WLAN interface.  Only called if not overridden by
71        another supported device.
72        """
73        raise NotImplementedError("{} must be defined.".format(
74            inspect.currentframe().f_code.co_name))
75
76    def take_bug_report(self, test_name, begin_time):
77        """Base generic WLAN interface.  Only called if not overridden by
78        another supported device.
79        """
80        raise NotImplementedError("{} must be defined.".format(
81            inspect.currentframe().f_code.co_name))
82
83    def get_log(self, test_name, begin_time):
84        """Base generic WLAN interface.  Only called if not overridden by
85        another supported device.
86        """
87        raise NotImplementedError("{} must be defined.".format(
88            inspect.currentframe().f_code.co_name))
89
90    def turn_location_off_and_scan_toggle_off(self):
91        """Base generic WLAN interface.  Only called if not overridden by
92        another supported device.
93        """
94        raise NotImplementedError("{} must be defined.".format(
95            inspect.currentframe().f_code.co_name))
96
97    def associate(self,
98                  target_ssid,
99                  target_pwd=None,
100                  check_connectivity=True,
101                  hidden=False):
102        """Base generic WLAN interface.  Only called if not overriden by
103        another supported device.
104        """
105        raise NotImplementedError("{} must be defined.".format(
106            inspect.currentframe().f_code.co_name))
107
108    def disconnect(self):
109        """Base generic WLAN interface.  Only called if not overridden by
110        another supported device.
111        """
112        raise NotImplementedError("{} must be defined.".format(
113            inspect.currentframe().f_code.co_name))
114
115    def get_wlan_interface_id_list(self):
116        """Base generic WLAN interface.  Only called if not overridden by
117        another supported device.
118        """
119        raise NotImplementedError("{} must be defined.".format(
120            inspect.currentframe().f_code.co_name))
121
122    def destroy_wlan_interface(self, iface_id):
123        """Base generic WLAN interface.  Only called if not overridden by
124        another supported device.
125        """
126        raise NotImplementedError("{} must be defined.".format(
127            inspect.currentframe().f_code.co_name))
128
129    def send_command(self, command):
130        raise NotImplementedError("{} must be defined.".format(
131            inspect.currentframe().f_code.co_name))
132
133    def get_interface_ip_addresses(self, interface):
134        raise NotImplementedError("{} must be defined.".format(
135            inspect.currentframe().f_code.co_name))
136
137    def is_connected(self, ssid=None):
138        raise NotImplementedError("{} must be defined.".format(
139            inspect.currentframe().f_code.co_name))
140
141    def ping(self, dest_ip, count=3, interval=1000, timeout=1000, size=25):
142        raise NotImplementedError("{} must be defined.".format(
143            inspect.currentframe().f_code.co_name))
144
145    def hard_power_cycle(self, pdus=None):
146        raise NotImplementedError("{} must be defined.".format(
147            inspect.currentframe().f_code.co_name))
148
149    def save_network(self, ssid):
150        raise NotImplementedError("{} must be defined.".format(
151            inspect.currentframe().f_code.co_name))
152
153    def clear_saved_networks(self):
154        raise NotImplementedError("{} must be defined.".format(
155            inspect.currentframe().f_code.co_name))
156
157
158class AndroidWlanDevice(WlanDevice):
159    """Class wrapper for an Android WLAN device.
160
161    Each object of this class represents a generic WLAN device.
162    Android device and Fuchsia devices are the currently supported devices/
163
164    Attributes:
165        android_device: An Android WLAN device.
166    """
167    def __init__(self, android_device):
168        super().__init__(android_device)
169
170    def wifi_toggle_state(self, state):
171        awutils.wifi_toggle_state(self.device, state)
172
173    def reset_wifi(self):
174        awutils.reset_wifi(self.device)
175
176    def take_bug_report(self, test_name, begin_time):
177        self.device.take_bug_report(test_name, begin_time)
178
179    def get_log(self, test_name, begin_time):
180        self.device.cat_adb_log(test_name, begin_time)
181
182    def turn_location_off_and_scan_toggle_off(self):
183        awutils.turn_location_off_and_scan_toggle_off(self.device)
184
185    def associate(self,
186                  target_ssid,
187                  target_pwd=None,
188                  key_mgmt=None,
189                  check_connectivity=True,
190                  hidden=False):
191        """Function to associate an Android WLAN device.
192
193        Args:
194            target_ssid: SSID to associate to.
195            target_pwd: Password for the SSID, if necessary.
196            key_mgmt: The hostapd wpa_key_mgmt value, distinguishes wpa3 from
197                wpa2 for android tests.
198            check_connectivity: Whether to check for internet connectivity.
199            hidden: Whether the network is hidden.
200        Returns:
201            True if successfully connected to WLAN, False if not.
202        """
203        network = {'SSID': target_ssid, 'hiddenSSID': hidden}
204        if target_pwd:
205            network['password'] = target_pwd
206        if key_mgmt:
207            network['security'] = key_mgmt
208        try:
209            awutils.connect_to_wifi_network(
210                self.device,
211                network,
212                check_connectivity=check_connectivity,
213                hidden=hidden)
214            return True
215        except Exception as e:
216            self.device.log.info('Failed to associated (%s)' % e)
217            return False
218
219    def disconnect(self):
220        awutils.turn_location_off_and_scan_toggle_off(self.device)
221
222    def get_wlan_interface_id_list(self):
223        pass
224
225    def destroy_wlan_interface(self, iface_id):
226        pass
227
228    def send_command(self, command):
229        return self.device.adb.shell(str(command))
230
231    def get_interface_ip_addresses(self, interface):
232        return get_interface_ip_addresses(self.device, interface)
233
234    def is_connected(self, ssid=None):
235        wifi_info = self.device.droid.wifiGetConnectionInfo()
236        if ssid:
237            return 'BSSID' in wifi_info and wifi_info['SSID'] == ssid
238        return 'BSSID' in wifi_info
239
240    def ping(self, dest_ip, count=3, interval=1000, timeout=1000, size=25):
241        return adb_shell_ping(self.device,
242                              dest_ip=dest_ip,
243                              count=count,
244                              timeout=timeout)
245
246    def hard_power_cycle(self, pdus):
247        pass
248
249    def save_network(self, ssid):
250        pass
251
252    def clear_saved_networks(self):
253        pass
254
255
256class FuchsiaWlanDevice(WlanDevice):
257    """Class wrapper for an Fuchsia WLAN device.
258
259    Each object of this class represents a generic WLAN device.
260    Android device and Fuchsia devices are the currently supported devices/
261
262    Attributes:
263        fuchsia_device: A Fuchsia WLAN device.
264    """
265    def __init__(self, fuchsia_device):
266        super().__init__(fuchsia_device)
267
268    def wifi_toggle_state(self, state):
269        """Stub for Fuchsia implementation."""
270        pass
271
272    def reset_wifi(self):
273        """Stub for Fuchsia implementation."""
274        pass
275
276    def take_bug_report(self, test_name, begin_time):
277        """Stub for Fuchsia implementation."""
278        pass
279
280    def get_log(self, test_name, begin_time):
281        """Stub for Fuchsia implementation."""
282        pass
283
284    def turn_location_off_and_scan_toggle_off(self):
285        """Stub for Fuchsia implementation."""
286        pass
287
288    def associate(self,
289                  target_ssid,
290                  target_pwd=None,
291                  key_mgmt=None,
292                  check_connectivity=True,
293                  hidden=False):
294        """Function to associate a Fuchsia WLAN device.
295
296        Args:
297            target_ssid: SSID to associate to.
298            target_pwd: Password for the SSID, if necessary.
299            key_mgmt: the hostapd wpa_key_mgmt, if specified.
300            check_connectivity: Whether to check for internet connectivity.
301            hidden: Whether the network is hidden.
302        Returns:
303            True if successfully connected to WLAN, False if not.
304        """
305        connection_response = self.device.wlan_lib.wlanConnectToNetwork(
306            target_ssid, target_pwd=target_pwd)
307
308        return self.device.check_connect_response(connection_response)
309
310    def disconnect(self):
311        """Function to disconnect from a Fuchsia WLAN device.
312           Asserts if disconnect was not successful.
313        """
314        disconnect_response = self.device.wlan_lib.wlanDisconnect()
315        asserts.assert_true(
316            self.device.check_disconnect_response(disconnect_response),
317            'Failed to disconnect.')
318
319    def status(self):
320        return self.device.wlan_lib.wlanStatus()
321
322    def ping(self, dest_ip, count=3, interval=1000, timeout=1000, size=25):
323        ping_result = self.device.ping(dest_ip,
324                                       count=count,
325                                       interval=interval,
326                                       timeout=timeout,
327                                       size=size)
328        return ping_result['status']
329
330    def get_wlan_interface_id_list(self):
331        """Function to list available WLAN interfaces.
332
333        Returns:
334            A list of wlan interface IDs.
335        """
336        return self.device.wlan_lib.wlanGetIfaceIdList().get('result')
337
338    def destroy_wlan_interface(self, iface_id):
339        """Function to associate a Fuchsia WLAN device.
340
341        Args:
342            target_ssid: SSID to associate to.
343            target_pwd: Password for the SSID, if necessary.
344            check_connectivity: Whether to check for internet connectivity.
345            hidden: Whether the network is hidden.
346        Returns:
347            True if successfully destroyed wlan interface, False if not.
348        """
349        result = self.device.wlan_lib.wlanDestroyIface(iface_id)
350        if result.get('error') is None:
351            return True
352        else:
353            self.log.error("Failed to destroy interface with: {}".format(
354                result.get('error')))
355            return False
356
357    def send_command(self, command):
358        return self.device.send_command_ssh(str(command)).stdout
359
360    def get_interface_ip_addresses(self, interface):
361        return get_interface_ip_addresses(self.device, interface)
362
363    def is_connected(self, ssid=None):
364        """ Determines if wlan_device is connected to wlan network.
365
366        Args:
367            ssid (optional): string, to check if device is connect to a specific
368                network.
369
370        Returns:
371            True, if connected to a network or to the correct network when SSID
372                is provided.
373            False, if not connected or connect to incorrect network when SSID is
374                provided.
375        """
376        response = self.status()
377        if response.get('error'):
378            raise ConnectionError(
379                'Failed to get client network connection status')
380
381        status = response.get('result')
382        if status and status.get('connected_to'):
383            if ssid:
384                connected_ssid = ''.join(
385                    chr(i) for i in status['connected_to']['ssid'])
386                if ssid != connected_ssid:
387                    return False
388            return True
389        return False
390
391    def hard_power_cycle(self, pdus):
392        self.device.reboot(reboot_type='hard', testbed_pdus=pdus)
393
394    def save_network(self, target_ssid, security_type=None, target_pwd=None):
395        if security_type and security_type not in FUCHSIA_VALID_SECURITY_TYPES:
396            raise TypeError('Invalid security type: %s' % security_type)
397        response = self.device.wlan_policy_lib.wlanSaveNetwork(
398            target_ssid, security_type, target_pwd=target_pwd)
399        if response.get('error'):
400            raise EnvironmentError('Failed to save network %s. Err: %s' %
401                                   (target_ssid, response.get('error')))
402
403    def clear_saved_networks(self):
404        # TODO(fxb/55852): Replace with WlanPolicyLib command once its implemented.
405        response = self.device.wlan_policy_lib.wlanGetSavedNetworks()
406        if response.get('error'):
407            raise ConnectionError('Failed to get saved networks: %s' %
408                                  response.get('error'))
409        for ssid in response.get('result'):
410            delete_response = self.device.wlan_policy_lib.wlanRemoveNetwork(
411                ssid, None)
412            if delete_response.get('error'):
413                raise EnvironmentError('Failed to delete network %s: %s' %
414                                       (ssid, delete_response.get('error')))
415