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
19import time
20
21import acts_contrib.test_utils.wifi.wifi_test_utils as awutils
22from acts.utils import adb_shell_ping
23
24from acts import asserts
25from acts.controllers import iperf_client
26from acts.controllers.fuchsia_device import FuchsiaDevice
27from acts.controllers.android_device import AndroidDevice
28
29FUCHSIA_VALID_SECURITY_TYPES = {"none", "wep", "wpa", "wpa2", "wpa3"}
30
31
32def create_wlan_device(hardware_device):
33    """Creates a generic WLAN device based on type of device that is sent to
34    the functions.
35
36    Args:
37        hardware_device: A WLAN hardware device that is supported by ACTS.
38    """
39    if isinstance(hardware_device, FuchsiaDevice):
40        return FuchsiaWlanDevice(hardware_device)
41    elif isinstance(hardware_device, AndroidDevice):
42        return AndroidWlanDevice(hardware_device)
43    else:
44        raise ValueError('Unable to create WlanDevice for type %s' %
45                         type(hardware_device))
46
47
48class WlanDevice(object):
49    """Class representing a generic WLAN device.
50
51    Each object of this class represents a generic WLAN device.
52    Android device and Fuchsia devices are the currently supported devices/
53
54    Attributes:
55        device: A generic WLAN device.
56    """
57
58    def __init__(self, device):
59        self.device = device
60        self.log = logging
61        self.identifier = None
62
63    def wifi_toggle_state(self, state):
64        """Base generic WLAN interface.  Only called if not overridden by
65        another supported device.
66        """
67        raise NotImplementedError("{} must be defined.".format(
68            inspect.currentframe().f_code.co_name))
69
70    def reset_wifi(self):
71        """Base generic WLAN interface.  Only called if not overridden by
72        another supported device.
73        """
74        raise NotImplementedError("{} must be defined.".format(
75            inspect.currentframe().f_code.co_name))
76
77    def take_bug_report(self, test_name=None, begin_time=None):
78        """Base generic WLAN interface.  Only called if not overridden by
79        another supported device.
80        """
81        raise NotImplementedError("{} must be defined.".format(
82            inspect.currentframe().f_code.co_name))
83
84    def get_log(self, test_name, begin_time):
85        """Base generic WLAN interface.  Only called if not overridden by
86        another supported device.
87        """
88        raise NotImplementedError("{} must be defined.".format(
89            inspect.currentframe().f_code.co_name))
90
91    def turn_location_off_and_scan_toggle_off(self):
92        """Base generic WLAN interface.  Only called if not overridden by
93        another supported device.
94        """
95        raise NotImplementedError("{} must be defined.".format(
96            inspect.currentframe().f_code.co_name))
97
98    def associate(self,
99                  target_ssid,
100                  target_pwd=None,
101                  check_connectivity=True,
102                  hidden=False,
103                  target_security=None):
104        """Base generic WLAN interface.  Only called if not overriden by
105        another supported device.
106        """
107        raise NotImplementedError("{} must be defined.".format(
108            inspect.currentframe().f_code.co_name))
109
110    def disconnect(self):
111        """Base generic WLAN interface.  Only called if not overridden by
112        another supported device.
113        """
114        raise NotImplementedError("{} must be defined.".format(
115            inspect.currentframe().f_code.co_name))
116
117    def get_wlan_interface_id_list(self):
118        """Base generic WLAN interface.  Only called if not overridden by
119        another supported device.
120        """
121        raise NotImplementedError("{} must be defined.".format(
122            inspect.currentframe().f_code.co_name))
123
124    def get_default_wlan_test_interface(self):
125        raise NotImplementedError("{} must be defined.".format(
126            inspect.currentframe().f_code.co_name))
127
128    def destroy_wlan_interface(self, iface_id):
129        """Base generic WLAN interface.  Only called if not overridden by
130        another supported device.
131        """
132        raise NotImplementedError("{} must be defined.".format(
133            inspect.currentframe().f_code.co_name))
134
135    def send_command(self, command):
136        raise NotImplementedError("{} must be defined.".format(
137            inspect.currentframe().f_code.co_name))
138
139    def is_connected(self, ssid=None):
140        raise NotImplementedError("{} must be defined.".format(
141            inspect.currentframe().f_code.co_name))
142
143    def can_ping(self,
144                 dest_ip,
145                 count=3,
146                 interval=1000,
147                 timeout=1000,
148                 size=25,
149                 additional_ping_params=None):
150        raise NotImplementedError("{} must be defined.".format(
151            inspect.currentframe().f_code.co_name))
152
153    def ping(self,
154             dest_ip,
155             count=3,
156             interval=1000,
157             timeout=1000,
158             size=25,
159             additional_ping_params=None):
160        raise NotImplementedError("{} must be defined.".format(
161            inspect.currentframe().f_code.co_name))
162
163    def hard_power_cycle(self, pdus=None):
164        raise NotImplementedError("{} must be defined.".format(
165            inspect.currentframe().f_code.co_name))
166
167    def save_network(self, ssid):
168        raise NotImplementedError("{} must be defined.".format(
169            inspect.currentframe().f_code.co_name))
170
171    def clear_saved_networks(self):
172        raise NotImplementedError("{} must be defined.".format(
173            inspect.currentframe().f_code.co_name))
174
175    def create_iperf_client(self, test_interface=None):
176        raise NotImplementedError("{} must be defined.".format(
177            inspect.currentframe().f_code.co_name))
178
179
180class AndroidWlanDevice(WlanDevice):
181    """Class wrapper for an Android WLAN device.
182
183    Each object of this class represents a generic WLAN device.
184    Android device and Fuchsia devices are the currently supported devices/
185
186    Attributes:
187        android_device: An Android WLAN device.
188    """
189
190    def __init__(self, android_device):
191        super().__init__(android_device)
192        self.identifier = android_device.serial
193
194    def wifi_toggle_state(self, state):
195        awutils.wifi_toggle_state(self.device, state)
196
197    def reset_wifi(self):
198        awutils.reset_wifi(self.device)
199
200    def take_bug_report(self, test_name=None, begin_time=None):
201        self.device.take_bug_report(test_name, begin_time)
202
203    def get_log(self, test_name, begin_time):
204        self.device.cat_adb_log(test_name, begin_time)
205
206    def turn_location_off_and_scan_toggle_off(self):
207        awutils.turn_location_off_and_scan_toggle_off(self.device)
208
209    def associate(self,
210                  target_ssid,
211                  target_pwd=None,
212                  key_mgmt=None,
213                  check_connectivity=True,
214                  hidden=False,
215                  target_security=None):
216        """Function to associate an Android WLAN device.
217
218        Args:
219            target_ssid: SSID to associate to.
220            target_pwd: Password for the SSID, if necessary.
221            key_mgmt: The hostapd wpa_key_mgmt value, distinguishes wpa3 from
222                wpa2 for android tests.
223            check_connectivity: Whether to check for internet connectivity.
224            hidden: Whether the network is hidden.
225        Returns:
226            True if successfully connected to WLAN, False if not.
227        """
228        network = {'SSID': target_ssid, 'hiddenSSID': hidden}
229        if target_pwd:
230            network['password'] = target_pwd
231        if key_mgmt:
232            network['security'] = key_mgmt
233        try:
234            awutils.connect_to_wifi_network(
235                self.device,
236                network,
237                check_connectivity=check_connectivity,
238                hidden=hidden)
239            return True
240        except Exception as e:
241            self.device.log.info('Failed to associated (%s)' % e)
242            return False
243
244    def disconnect(self):
245        awutils.turn_location_off_and_scan_toggle_off(self.device)
246
247    def get_wlan_interface_id_list(self):
248        pass
249
250    def get_default_wlan_test_interface(self):
251        return 'wlan0'
252
253    def destroy_wlan_interface(self, iface_id):
254        pass
255
256    def send_command(self, command):
257        return self.device.adb.shell(str(command))
258
259    def is_connected(self, ssid=None):
260        wifi_info = self.device.droid.wifiGetConnectionInfo()
261        if ssid:
262            return 'BSSID' in wifi_info and wifi_info['SSID'] == ssid
263        return 'BSSID' in wifi_info
264
265    def can_ping(self,
266                 dest_ip,
267                 count=3,
268                 interval=1000,
269                 timeout=1000,
270                 size=25,
271                 additional_ping_params=None):
272        return adb_shell_ping(self.device,
273                              dest_ip=dest_ip,
274                              count=count,
275                              timeout=timeout)
276
277    def ping(self, dest_ip, count=3, interval=1000, timeout=1000, size=25):
278        pass
279
280    def hard_power_cycle(self, pdus):
281        pass
282
283    def save_network(self, ssid):
284        pass
285
286    def clear_saved_networks(self):
287        pass
288
289    def create_iperf_client(self, test_interface=None):
290        """ Returns an iperf client on the Android, without requiring a
291        specific config.
292
293        Args:
294            test_interface: optional, string, name of test interface.
295
296        Returns:
297            IPerfClient object
298        """
299        if not test_interface:
300            test_interface = self.get_default_wlan_test_interface()
301
302        return iperf_client.IPerfClientOverAdb(
303            android_device_or_serial=self.device,
304            test_interface=test_interface)
305
306
307class FuchsiaWlanDevice(WlanDevice):
308    """Class wrapper for an Fuchsia WLAN device.
309
310    Each object of this class represents a generic WLAN device.
311    Android device and Fuchsia devices are the currently supported devices/
312
313    Attributes:
314        fuchsia_device: A Fuchsia WLAN device.
315    """
316
317    def __init__(self, fuchsia_device):
318        super().__init__(fuchsia_device)
319        self.identifier = fuchsia_device.ip
320        self.device.configure_wlan()
321
322    def wifi_toggle_state(self, state):
323        """Stub for Fuchsia implementation."""
324
325    def reset_wifi(self):
326        """Stub for Fuchsia implementation."""
327
328    def take_bug_report(self, test_name=None, begin_time=None):
329        """Stub for Fuchsia implementation."""
330        self.device.take_bug_report(test_name, begin_time)
331
332    def get_log(self, test_name, begin_time):
333        """Stub for Fuchsia implementation."""
334
335    def turn_location_off_and_scan_toggle_off(self):
336        """Stub for Fuchsia implementation."""
337
338    def associate(self,
339                  target_ssid,
340                  target_pwd=None,
341                  key_mgmt=None,
342                  check_connectivity=True,
343                  hidden=False,
344                  target_security=None):
345        """Function to associate a Fuchsia WLAN device.
346
347        Args:
348            target_ssid: SSID to associate to.
349            target_pwd: Password for the SSID, if necessary.
350            key_mgmt: the hostapd wpa_key_mgmt, if specified.
351            check_connectivity: Whether to check for internet connectivity.
352            hidden: Whether the network is hidden.
353            target_security: string, target security for network, used to
354                save the network in policy connects (see wlan_policy_lib)
355        Returns:
356            True if successfully connected to WLAN, False if not.
357        """
358        if self.device.association_mechanism == 'drivers':
359            bss_scan_response = self.device.sl4f.wlan_lib.wlanScanForBSSInfo()
360            if bss_scan_response.get('error'):
361                self.log.error('Scan for BSS info failed. Err: %s' %
362                               bss_scan_response['error'])
363                return False
364
365            bss_descs_for_ssid = bss_scan_response['result'].get(
366                target_ssid, None)
367            if not bss_descs_for_ssid or len(bss_descs_for_ssid) < 1:
368                self.log.error(
369                    'Scan failed to find a BSS description for target_ssid %s'
370                    % target_ssid)
371                return False
372
373            connection_response = self.device.sl4f.wlan_lib.wlanConnectToNetwork(
374                target_ssid, bss_descs_for_ssid[0], target_pwd=target_pwd)
375            return self.device.check_connect_response(connection_response)
376        else:
377            return self.device.wlan_policy_controller.save_and_connect(
378                target_ssid, target_security, password=target_pwd)
379
380    def disconnect(self):
381        """Function to disconnect from a Fuchsia WLAN device.
382           Asserts if disconnect was not successful.
383        """
384        if self.device.association_mechanism == 'drivers':
385            disconnect_response = self.device.sl4f.wlan_lib.wlanDisconnect()
386            return self.device.check_disconnect_response(disconnect_response)
387        else:
388            return self.device.wlan_policy_controller.remove_all_networks_and_wait_for_no_connections(
389            )
390
391    def status(self):
392        return self.device.sl4f.wlan_lib.wlanStatus()
393
394    def can_ping(self,
395                 dest_ip,
396                 count=3,
397                 interval=1000,
398                 timeout=1000,
399                 size=25,
400                 additional_ping_params=None):
401        return self.device.can_ping(
402            dest_ip,
403            count=count,
404            interval=interval,
405            timeout=timeout,
406            size=size,
407            additional_ping_params=additional_ping_params)
408
409    def ping(self,
410             dest_ip,
411             count=3,
412             interval=1000,
413             timeout=1000,
414             size=25,
415             additional_ping_params=None):
416        return self.device.ping(dest_ip,
417                                count=count,
418                                interval=interval,
419                                timeout=timeout,
420                                size=size,
421                                additional_ping_params=additional_ping_params)
422
423    def get_wlan_interface_id_list(self):
424        """Function to list available WLAN interfaces.
425
426        Returns:
427            A list of wlan interface IDs.
428        """
429        return self.device.sl4f.wlan_lib.wlanGetIfaceIdList().get('result')
430
431    def get_default_wlan_test_interface(self):
432        """Returns name of the WLAN client interface"""
433        return self.device.wlan_client_test_interface_name
434
435    def destroy_wlan_interface(self, iface_id):
436        """Function to associate a Fuchsia WLAN device.
437
438        Args:
439            target_ssid: SSID to associate to.
440            target_pwd: Password for the SSID, if necessary.
441            check_connectivity: Whether to check for internet connectivity.
442            hidden: Whether the network is hidden.
443        Returns:
444            True if successfully destroyed wlan interface, False if not.
445        """
446        result = self.device.sl4f.wlan_lib.wlanDestroyIface(iface_id)
447        if result.get('error') is None:
448            return True
449        else:
450            self.log.error("Failed to destroy interface with: {}".format(
451                result.get('error')))
452            return False
453
454    def send_command(self, command):
455        return self.device.ssh.run(str(command)).stdout
456
457    def is_connected(self, ssid=None):
458        """ Determines if wlan_device is connected to wlan network.
459
460        Args:
461            ssid (optional): string, to check if device is connect to a specific
462                network.
463
464        Returns:
465            True, if connected to a network or to the correct network when SSID
466                is provided.
467            False, if not connected or connect to incorrect network when SSID is
468                provided.
469        """
470        response = self.status()
471        if response.get('error'):
472            raise ConnectionError(
473                'Failed to get client network connection status')
474        result = response.get('result')
475        if isinstance(result, dict):
476            connected_to = result.get('Connected')
477            # TODO(https://fxbug.dev/85938): Remove backwards compatibility once
478            # ACTS is versioned with Fuchsia.
479            if not connected_to:
480                connected_to = result.get('connected_to')
481            if not connected_to:
482                return False
483
484            if ssid:
485                # Replace encoding errors instead of raising an exception.
486                # Since `ssid` is a string, this will not affect the test
487                # for equality.
488                connected_ssid = bytearray(connected_to['ssid']).decode(
489                    encoding='utf-8', errors='replace')
490                return ssid == connected_ssid
491            return True
492        return False
493
494    def hard_power_cycle(self, pdus):
495        self.device.reboot(reboot_type='hard', testbed_pdus=pdus)
496
497    def save_network(self, target_ssid, security_type=None, target_pwd=None):
498        if self.device.association_mechanism == 'drivers':
499            raise EnvironmentError(
500                'Cannot save network using the drivers. Saved networks are a '
501                'policy layer concept.')
502        if security_type and security_type not in FUCHSIA_VALID_SECURITY_TYPES:
503            raise TypeError('Invalid security type: %s' % security_type)
504        if not self.device.wlan_policy_controller.save_network(
505                target_ssid, security_type, password=target_pwd):
506            raise EnvironmentError('Failed to save network: %s' % target_ssid)
507
508    def clear_saved_networks(self):
509        if self.device.association_mechanism == 'drivers':
510            raise EnvironmentError(
511                'Cannot clear saved network using the drivers. Saved networks '
512                'are a policy layer concept.')
513        if not self.device.wlan_policy_controller.remove_all_networks():
514            raise EnvironmentError('Failed to clear saved networks')
515
516    def create_iperf_client(self, test_interface=None):
517        """ Returns an iperf client on the FuchsiaDevice, without requiring a
518        specific config.
519
520        Args:
521            test_interface: optional, string, name of test interface. Defaults
522                to first found wlan client interface.
523
524        Returns:
525            IPerfClient object
526        """
527        if not test_interface:
528            test_interface = self.get_default_wlan_test_interface()
529
530        # A package server is necessary to acquire the iperf3 client for
531        # some builds.
532        self.device.start_package_server()
533
534        return iperf_client.IPerfClientOverSsh(
535            {
536                'user': 'fuchsia',
537                'host': self.device.ip,
538                'ssh_config': self.device.ssh_config
539            },
540            ssh_provider=self.device.ssh,
541            test_interface=test_interface)
542