android-13.0.0_r83/s

#!/usr/bin/env python3
#
#   Copyright 2020 - The Android Open Source Project
#
#   Licensed under the Apache License, Version 2.0 (the 'License');
#   you may not use this file except in compliance with the License.
#   You may obtain a copy of the License at
#
#       http://www.apache.org/licenses/LICENSE-2.0
#
#   Unless required by applicable law or agreed to in writing, software
#   distributed under the License is distributed on an 'AS IS' BASIS,
#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#   See the License for the specific language governing permissions and
#   limitations under the License.

import collections.abc
import copy
import fcntl
import importlib
import os
import selenium
import splinter
import time
from acts import logger

BROWSER_WAIT_SHORT = 1
BROWSER_WAIT_MED = 3
BROWSER_WAIT_LONG = 30
BROWSER_WAIT_EXTRA_LONG = 60


def create(configs):
    """Factory method for retail AP class.

    Args:
        configs: list of dicts containing ap settings. ap settings must contain
        the following: brand, model, ip_address, username and password
    """
    SUPPORTED_APS = {
        ('Netgear', 'R7000'): {
            'name': 'NetgearR7000AP',
            'package': 'netgear_r7000'
        },
        ('Netgear', 'R7000NA'): {
            'name': 'NetgearR7000NAAP',
            'package': 'netgear_r7000'
        },
        ('Netgear', 'R7500'): {
            'name': 'NetgearR7500AP',
            'package': 'netgear_r7500'
        },
        ('Netgear', 'R7500NA'): {
            'name': 'NetgearR7500NAAP',
            'package': 'netgear_r7500'
        },
        ('Netgear', 'R7800'): {
            'name': 'NetgearR7800AP',
            'package': 'netgear_r7800'
        },
        ('Netgear', 'R8000'): {
            'name': 'NetgearR8000AP',
            'package': 'netgear_r8000'
        },
        ('Netgear', 'RAX80'): {
            'name': 'NetgearRAX80AP',
            'package': 'netgear_rax80'
        },
        ('Netgear', 'RAX120'): {
            'name': 'NetgearRAX120AP',
            'package': 'netgear_rax120'
        },
        ('Netgear', 'RAX200'): {
            'name': 'NetgearRAX200AP',
            'package': 'netgear_rax200'
        },
        ('Netgear', 'RAXE500'): {
            'name': 'NetgearRAXE500AP',
            'package': 'netgear_raxe500'
        },
        ('Brcm', 'Reference'): {
            'name': 'BrcmRefAP',
            'package': 'brcm_ref'
        },
        ('Google', 'Wifi'): {
            'name': 'GoogleWifiAP',
            'package': 'google_wifi'
        },
    }
    objs = []
    for config in configs:
        ap_id = (config['brand'], config['model'])
        if ap_id not in SUPPORTED_APS:
            raise KeyError('Invalid retail AP brand and model combination.')
        ap_class_dict = SUPPORTED_APS[ap_id]
        ap_package = 'acts_contrib.test_utils.wifi.wifi_retail_ap.{}'.format(
            ap_class_dict['package'])
        ap_package = importlib.import_module(ap_package)
        ap_class = getattr(ap_package, ap_class_dict['name'])
        objs.append(ap_class(config))
    return objs


def destroy(objs):
    for obj in objs:
        obj.teardown()


class BlockingBrowser(splinter.driver.webdriver.chrome.WebDriver):
    """Class that implements a blocking browser session on top of selenium.

    The class inherits from and builds upon splinter/selenium's webdriver class
    and makes sure that only one such webdriver is active on a machine at any
    single time. The class ensures single session operation using a lock file.
    The class is to be used within context managers (e.g. with statements) to
    ensure locks are always properly released.
    """
    def __init__(self, headless, timeout):
        """Constructor for BlockingBrowser class.

        Args:
            headless: boolean to control visible/headless browser operation
            timeout: maximum time allowed to launch browser
        """
        self.log = logger.create_tagged_trace_logger('ChromeDriver')
        self.chrome_options = splinter.driver.webdriver.chrome.Options()
        self.chrome_options.add_argument('--no-proxy-server')
        self.chrome_options.add_argument('--no-sandbox')
        self.chrome_options.add_argument('--allow-running-insecure-content')
        self.chrome_options.add_argument('--ignore-certificate-errors')
        self.chrome_capabilities = selenium.webdriver.common.desired_capabilities.DesiredCapabilities.CHROME.copy(
        )
        self.chrome_capabilities['acceptSslCerts'] = True
        self.chrome_capabilities['acceptInsecureCerts'] = True
        if headless:
            self.chrome_options.add_argument('--headless')
            self.chrome_options.add_argument('--disable-gpu')
        self.lock_file_path = '/usr/local/bin/chromedriver'
        self.timeout = timeout

    def __enter__(self):
        """Entry context manager for BlockingBrowser.

        The enter context manager for BlockingBrowser attempts to lock the
        browser file. If successful, it launches and returns a chromedriver
        session. If an exception occurs while starting the browser, the lock
        file is released.
        """
        self.lock_file = open(self.lock_file_path, 'r')
        start_time = time.time()
        while time.time() < start_time + self.timeout:
            try:
                fcntl.flock(self.lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
            except BlockingIOError:
                time.sleep(BROWSER_WAIT_SHORT)
                continue
            try:
                self.driver = selenium.webdriver.Chrome(
                    options=self.chrome_options,
                    desired_capabilities=self.chrome_capabilities)
                self.element_class = splinter.driver.webdriver.WebDriverElement
                self._cookie_manager = splinter.driver.webdriver.cookie_manager.CookieManager(
                    self.driver)
                super(splinter.driver.webdriver.chrome.WebDriver,
                      self).__init__(2)
                return super(BlockingBrowser, self).__enter__()
            except:
                fcntl.flock(self.lock_file, fcntl.LOCK_UN)
                self.lock_file.close()
                raise RuntimeError('Error starting browser. '
                                   'Releasing lock file.')
        raise TimeoutError('Could not start chrome browser in time.')

    def __exit__(self, exc_type, exc_value, traceback):
        """Exit context manager for BlockingBrowser.

        The exit context manager simply calls the parent class exit and
        releases the lock file.
        """
        try:
            super(BlockingBrowser, self).__exit__(exc_type, exc_value,
                                                  traceback)
        except:
            raise RuntimeError('Failed to quit browser. Releasing lock file.')
        finally:
            fcntl.flock(self.lock_file, fcntl.LOCK_UN)
            self.lock_file.close()

    def restart(self):
        """Method to restart browser session without releasing lock file."""
        self.quit()
        self.__enter__()

    def visit_persistent(self,
                         url,
                         page_load_timeout,
                         num_tries,
                         backup_url='about:blank',
                         check_for_element=None):
        """Method to visit webpages and retry upon failure.

        The function visits a URL and checks that the resulting URL matches
        the intended URL, i.e. no redirects have happened

        Args:
            url: the intended url
            page_load_timeout: timeout for page visits
            num_tries: number of tries before url is declared unreachable
            backup_url: url to visit if first url is not reachable. This can be
            used to simply refresh the browser and try again or to re-login to
            the AP
            check_for_element: element id to check for existence on page
        """
        self.driver.set_page_load_timeout(page_load_timeout)
        for idx in range(num_tries):
            try:
                self.visit(url)
            except:
                self.restart()

            page_reached = self.url.split('/')[-1] == url.split('/')[-1]
            if check_for_element:
                time.sleep(BROWSER_WAIT_MED)
                element = self.find_by_id(check_for_element)
                if not element:
                    page_reached = 0
            if page_reached:
                break
            else:
                try:
                    self.visit(backup_url)
                except:
                    self.restart()

            if idx == num_tries - 1:
                self.log.error('URL unreachable. Current URL: {}'.format(
                    self.url))
                raise RuntimeError('URL unreachable.')


class WifiRetailAP(object):
    """Base class implementation for retail ap.

    Base class provides functions whose implementation is shared by all aps.
    If some functions such as set_power not supported by ap, checks will raise
    exceptions.
    """
    def __init__(self, ap_settings):
        self.ap_settings = ap_settings.copy()
        self.log = logger.create_tagged_trace_logger('AccessPoint|{}'.format(
            self._get_control_ip_address()))
        # Capabilities variable describing AP capabilities
        self.capabilities = {
            'interfaces': [],
            'channels': {},
            'modes': {},
            'default_mode': None
        }
        for interface in self.capabilities['interfaces']:
            self.ap_settings.setdefault(interface, {})
        # Lock AP
        if self.ap_settings.get('lock_ap', 0):
            self.lock_timeout = self.ap_settings.get('lock_timeout', 3600)
            self._lock_ap()

    def teardown(self):
        """Function to perform destroy operations."""
        if self.ap_settings.get('lock_ap', 0):
            self._unlock_ap()

    def reset(self):
        """Function that resets AP.

        Function implementation is AP dependent and intended to perform any
        necessary reset operations as part of controller destroy.
        """
        pass

    def read_ap_settings(self):
        """Function that reads current ap settings.

        Function implementation is AP dependent and thus base class raises exception
        if function not implemented in child class.
        """
        raise NotImplementedError

    def validate_ap_settings(self):
        """Function to validate ap settings.

        This function compares the actual ap settings read from the web GUI
        with the assumed settings saved in the AP object. When called after AP
        configuration, this method helps ensure that our configuration was
        successful.
        Note: Calling this function updates the stored ap_settings

        Raises:
            ValueError: If read AP settings do not match stored settings.
        """
        assumed_ap_settings = copy.deepcopy(self.ap_settings)
        actual_ap_settings = self.read_ap_settings()

        if assumed_ap_settings != actual_ap_settings:
            self.log.warning(
                'Discrepancy in AP settings. Some settings may have been overwritten.'
            )

    def configure_ap(self, **config_flags):
        """Function that configures ap based on values of ap_settings.

        Function implementation is AP dependent and thus base class raises exception
        if function not implemented in child class.

        Args:
            config_flags: optional configuration flags
        """
        raise NotImplementedError

    def set_region(self, region):
        """Function that sets AP region.

        This function sets the region for the AP. Note that this may overwrite
        channel and bandwidth settings in cases where the new region does not
        support the current wireless configuration.

        Args:
            region: string indicating AP region
        """
        if region != self.ap_settings['region']:
            self.log.warning(
                'Updating region may overwrite wireless settings.')
        setting_to_update = {'region': region}
        self.update_ap_settings(setting_to_update)

    def set_radio_on_off(self, network, status):
        """Function that turns the radio on or off.

        Args:
            network: string containing network identifier (2G, 5G_1, 5G_2)
            status: boolean indicating on or off (0: off, 1: on)
        """
        setting_to_update = {network: {'status': int(status)}}
        self.update_ap_settings(setting_to_update)

    def set_ssid(self, network, ssid):
        """Function that sets network SSID.

        Args:
            network: string containing network identifier (2G, 5G_1, 5G_2)
            ssid: string containing ssid
        """
        setting_to_update = {network: {'ssid': str(ssid)}}
        self.update_ap_settings(setting_to_update)

    def set_channel(self, network, channel):
        """Function that sets network channel.

        Args:
            network: string containing network identifier (2G, 5G_1, 5G_2)
            channel: string or int containing channel
        """
        if channel not in self.capabilities['channels'][network]:
            self.log.error('Ch{} is not supported on {} interface.'.format(
                channel, network))
        setting_to_update = {network: {'channel': channel}}
        self.update_ap_settings(setting_to_update)

    def set_bandwidth(self, network, bandwidth):
        """Function that sets network bandwidth/mode.

        Args:
            network: string containing network identifier (2G, 5G_1, 5G_2)
            bandwidth: string containing mode, e.g. 11g, VHT20, VHT40, VHT80.
        """
        if 'bw' in bandwidth:
            bandwidth = bandwidth.replace('bw',
                                          self.capabilities['default_mode'])
        elif isinstance(bandwidth, int):
            bandwidth = str(bandwidth) + self.capabilities['default_mode']
        if bandwidth not in self.capabilities['modes'][network]:
            self.log.error('{} mode is not supported on {} interface.'.format(
                bandwidth, network))
        setting_to_update = {network: {'bandwidth': bandwidth}}
        self.update_ap_settings(setting_to_update)

    def set_channel_and_bandwidth(self, network, channel, bandwidth):
        """Function that sets network bandwidth/mode and channel.

        Args:
            network: string containing network identifier (2G, 5G_1, 5G_2)
            channel: string containing desired channel
            bandwidth: string containing mode, e.g. 11g, VHT20, VHT40, VHT80.
        """
        if 'bw' in bandwidth:
            bandwidth = bandwidth.replace('bw',
                                          self.capabilities['default_mode'])
        elif isinstance(bandwidth, int):
            bandwidth = str(bandwidth) + self.capabilities['default_mode']
        if bandwidth not in self.capabilities['modes'][network]:
            self.log.error('{} mode is not supported on {} interface.'.format(
                bandwidth, network))
        if channel not in self.capabilities['channels'][network]:
            self.log.error('Ch{} is not supported on {} interface.'.format(
                channel, network))
        setting_to_update = {
            network: {
                'bandwidth': bandwidth,
                'channel': channel
            }
        }
        self.update_ap_settings(setting_to_update)

    def set_power(self, network, power):
        """Function that sets network transmit power.

        Args:
            network: string containing network identifier (2G, 5G_1, 5G_2)
            power: string containing power level, e.g., 25%, 100%
        """
        if 'power' not in self.ap_settings[network].keys():
            self.log.error(
                'Cannot configure power on {} interface.'.format(network))
        setting_to_update = {network: {'power': power}}
        self.update_ap_settings(setting_to_update)

    def set_security(self, network, security_type, *password):
        """Function that sets network security setting and password.

        Args:
            network: string containing network identifier (2G, 5G_1, 5G_2)
            security: string containing security setting, e.g., WPA2-PSK
            password: optional argument containing password
        """
        if (len(password) == 1) and (type(password[0]) == str):
            setting_to_update = {
                network: {
                    'security_type': str(security_type),
                    'password': str(password[0])
                }
            }
        else:
            setting_to_update = {
                network: {
                    'security_type': str(security_type)
                }
            }
        self.update_ap_settings(setting_to_update)

    def set_rate(self):
        """Function that configures rate used by AP.

        Function implementation is not supported by most APs and thus base
        class raises exception if function not implemented in child class.
        """
        raise NotImplementedError

    def _update_settings_dict(self,
                              settings,
                              updates,
                              updates_requested=False,
                              status_toggle_flag=False):
        new_settings = copy.deepcopy(settings)
        for key, value in updates.items():
            if key not in new_settings.keys():
                raise KeyError('{} is an invalid settings key.'.format(key))
            elif isinstance(value, collections.abc.Mapping):
                new_settings[
                    key], updates_requested, status_toggle_flag = self._update_settings_dict(
                        new_settings.get(key, {}), value, updates_requested,
                        status_toggle_flag)
            elif new_settings[key] != value:
                new_settings[key] = value
                updates_requested = True
                if 'status' in key:
                    status_toggle_flag = True
        return new_settings, updates_requested, status_toggle_flag

    def update_ap_settings(self, dict_settings={}, **named_settings):
        """Function to update settings of existing AP.

        Function copies arguments into ap_settings and calls configure_retail_ap
        to apply them.

        Args:
            *dict_settings accepts single dictionary of settings to update
            **named_settings accepts named settings to update
            Note: dict and named_settings cannot contain the same settings.
        """
        settings_to_update = dict(dict_settings, **named_settings)
        if len(settings_to_update) != len(dict_settings) + len(named_settings):
            raise KeyError('The following keys were passed twice: {}'.format(
                (set(dict_settings.keys()).intersection(
                    set(named_settings.keys())))))

        self.ap_settings, updates_requested, status_toggle_flag = self._update_settings_dict(
            self.ap_settings, settings_to_update)

        if updates_requested:
            self.configure_ap(status_toggled=status_toggle_flag)

    def band_lookup_by_channel(self, channel):
        """Function that gives band name by channel number.

        Args:
            channel: channel number to lookup
        Returns:
            band: name of band which this channel belongs to on this ap, False
            if not supported
        """
        for key, value in self.capabilities['channels'].items():
            if channel in value:
                return key
        return False

    def _get_control_ip_address(self):
        """Function to get AP's Control Interface IP address."""
        if 'ssh_config' in self.ap_settings.keys():
            return self.ap_settings['ssh_config']['host']
        else:
            return self.ap_settings['ip_address']

    def _lock_ap(self):
        """Function to lock the ap while tests are running."""
        self.lock_file_path = '/tmp/{}_{}_{}.lock'.format(
            self.ap_settings['brand'], self.ap_settings['model'],
            self._get_control_ip_address())
        if not os.path.exists(self.lock_file_path):
            with open(self.lock_file_path, 'w'):
                pass
        self.lock_file = open(self.lock_file_path, 'r')
        start_time = time.time()
        self.log.info('Trying to acquire AP lock.')
        while time.time() < start_time + self.lock_timeout:
            try:
                fcntl.flock(self.lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
            except BlockingIOError:
                time.sleep(BROWSER_WAIT_SHORT)
                continue
            self.log.info('AP lock acquired.')
            return
        raise RuntimeError('Could not lock AP in time.')

    def _unlock_ap(self):
        """Function to unlock the AP when tests are done."""
        self.log.info('Releasing AP lock.')
        if hasattr(self, 'lock_file'):
            try:
                fcntl.flock(self.lock_file, fcntl.LOCK_UN)
                self.lock_file.close()
                self.log.info('Succussfully released AP lock file.')
            except:
                raise RuntimeError('Error occurred while unlocking AP.')