android-12.0.0_r34/s

# Lint as: python2, python3
# Copyright 2015 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""A module providing common resources for different facades."""

import exceptions
import logging
import time

from autotest_lib.client.bin import utils
from autotest_lib.client.common_lib.cros import chrome
from autotest_lib.client.common_lib.cros import retry
from autotest_lib.client.cros import constants

import py_utils

_FLAKY_CALL_RETRY_TIMEOUT_SEC = 60
_FLAKY_CHROME_CALL_RETRY_DELAY_SEC = 1

retry_chrome_call = retry.retry(
        (chrome.Error, exceptions.IndexError, exceptions.Exception),
        timeout_min=_FLAKY_CALL_RETRY_TIMEOUT_SEC / 60.0,
        delay_sec=_FLAKY_CHROME_CALL_RETRY_DELAY_SEC)


class FacadeResoureError(Exception):
    """Error in FacadeResource."""
    pass


_FLAKY_CHROME_START_RETRY_TIMEOUT_SEC = 120
_FLAKY_CHROME_START_RETRY_DELAY_SEC = 10


# Telemetry sometimes fails to start Chrome.
retry_start_chrome = retry.retry(
        (Exception,),
        timeout_min=_FLAKY_CHROME_START_RETRY_TIMEOUT_SEC / 60.0,
        delay_sec=_FLAKY_CHROME_START_RETRY_DELAY_SEC,
        exception_to_raise=FacadeResoureError,
        label='Start Chrome')


class FacadeResource(object):
    """This class provides access to telemetry chrome wrapper."""

    ARC_DISABLED = 'disabled'
    ARC_ENABLED = 'enabled'
    ARC_VERSION = 'CHROMEOS_ARC_VERSION'
    EXTRA_BROWSER_ARGS = ['--enable-gpu-benchmarking', '--use-fake-ui-for-media-stream']

    def __init__(self, chrome_object=None, restart=False):
        """Initializes a FacadeResource.

        @param chrome_object: A chrome.Chrome object or None.
        @param restart: Preserve the previous browser state.

        """
        self._chrome = chrome_object

    @property
    def _browser(self):
        """Gets the browser object from Chrome."""
        return self._chrome.browser


    @retry_start_chrome
    def _start_chrome(self, kwargs):
        """Start a Chrome with given arguments.

        @param kwargs: A dict of keyword arguments passed to Chrome.

        @return: A chrome.Chrome object.

        """
        logging.debug('Try to start Chrome with kwargs: %s', kwargs)
        return chrome.Chrome(**kwargs)


    def start_custom_chrome(self, kwargs):
        """Start a custom Chrome with given arguments.

        @param kwargs: A dict of keyword arguments passed to Chrome.

        @return: True on success, False otherwise.

        """
        # Close the previous Chrome.
        if self._chrome:
            self._chrome.close()

        # Start the new Chrome.
        try:
            self._chrome = self._start_chrome(kwargs)
        except FacadeResoureError:
            logging.error('Failed to start Chrome after retries')
            return False
        else:
            logging.info('Chrome started successfully')

        # The opened tabs are stored by tab descriptors.
        # Key is the tab descriptor string.
        # We use string as the key because of RPC Call. Client can use the
        # string to locate the tab object.
        # Value is the tab object.
        self._tabs = dict()

        # Workaround for issue crbug.com/588579.
        # On daisy, Chrome freezes about 30 seconds after login because of
        # TPM error. Avoid test accessing Chrome during this time.
        # Check issue crbug.com/588579 and crbug.com/591646.
        if utils.get_board() == 'daisy':
            logging.warning('Delay 30s for issue 588579 on daisy')
            time.sleep(30)

        return True


    def start_default_chrome(self, restart=False, extra_browser_args=None,
                             disable_arc=False):
        """Start the default Chrome.

        @param restart: True to start Chrome without clearing previous state.
        @param extra_browser_args: A list containing extra browser args passed
                                   to Chrome. This list will be appened to
                                   default EXTRA_BROWSER_ARGS.
        @param disable_arc: True to disable ARC++.

        @return: True on success, False otherwise.

        """
        # TODO: (crbug.com/618111) Add test driven switch for
        # supporting arc_mode enabled or disabled. At this time
        # if ARC build is tested, arc_mode is always enabled.
        if not disable_arc and utils.get_board_property(self.ARC_VERSION):
            arc_mode = self.ARC_ENABLED
        else:
            arc_mode = self.ARC_DISABLED
        kwargs = {
            'extension_paths': [constants.AUDIO_TEST_EXTENSION,
                                constants.DISPLAY_TEST_EXTENSION],
            'extra_browser_args': self.EXTRA_BROWSER_ARGS,
            'clear_enterprise_policy': not restart,
            'arc_mode': arc_mode,
            'autotest_ext': True
        }
        if extra_browser_args:
            kwargs['extra_browser_args'] += extra_browser_args
        return self.start_custom_chrome(kwargs)


    def __enter__(self):
        return self


    def __exit__(self, *args):
        if self._chrome:
            self._chrome.close()
            self._chrome = None


    @staticmethod
    def _generate_tab_descriptor(tab):
        """Generate tab descriptor by tab object.

        @param tab: the tab object.
        @return a str, the tab descriptor of the tab.

        """
        return hex(id(tab))


    def clean_unexpected_tabs(self):
        """Clean all tabs that are not opened by facade_resource

        It is used to make sure our chrome browser is clean.

        """
        # If they have the same length we can assume there is no unexpected
        # tabs.
        browser_tabs = self.get_tabs()
        if len(browser_tabs) == len(self._tabs):
            return

        for tab in browser_tabs:
            if self._generate_tab_descriptor(tab) not in self._tabs:
                # TODO(mojahsu): Reevaluate this code. crbug.com/719592
                try:
                    tab.Close()
                except py_utils.TimeoutException:
                    logging.warn('close tab timeout %r, %s', tab, tab.url)


    @retry_chrome_call
    def get_extension(self, extension_path=None):
        """Gets the extension from the indicated path.

        @param extension_path: the path of the target extension.
                               Set to None to get autotest extension.
                               Defaults to None.
        @return an extension object.

        @raise RuntimeError if the extension is not found.
        @raise chrome.Error if the found extension has not yet been
               retrieved succesfully.

        """
        try:
            if extension_path is None:
                extension = self._chrome.autotest_ext
            else:
                extension = self._chrome.get_extension(extension_path)
        except KeyError as errmsg:
            # Trigger retry_chrome_call to retry to retrieve the
            # found extension.
            raise chrome.Error(errmsg)
        if not extension:
            if extension_path is None:
                raise RuntimeError('Autotest extension not found')
            else:
                raise RuntimeError('Extension not found in %r'
                                    % extension_path)
        return extension


    def get_visible_notifications(self):
        """Gets the visible notifications

        @return: Returns all visible notifications in list format. Ex:
                [{title:'', message:'', prority:'', id:''}]
        """
        return self._chrome.get_visible_notifications()


    @retry_chrome_call
    def load_url(self, url):
        """Loads the given url in a new tab. The new tab will be active.

        @param url: The url to load as a string.
        @return a str, the tab descriptor of the opened tab.

        """
        tab = self._browser.tabs.New()
        tab.Navigate(url)
        tab.Activate()
        tab.WaitForDocumentReadyStateToBeComplete()
        tab_descriptor = self._generate_tab_descriptor(tab)
        self._tabs[tab_descriptor] = tab
        self.clean_unexpected_tabs()
        return tab_descriptor


    def set_http_server_directories(self, directories):
        """Starts an HTTP server.

        @param directories: Directories to start serving.

        @return True on success. False otherwise.

        """
        return self._chrome.browser.platform.SetHTTPServerDirectories(directories)


    def http_server_url_of(self, fullpath):
        """Converts a path to a URL.

        @param fullpath: String containing the full path to the content.

        @return the URL for the provided path.

        """
        return self._chrome.browser.platform.http_server.UrlOf(fullpath)


    def get_tabs(self):
        """Gets the tabs opened by browser.

        @returns: The tabs attribute in telemetry browser object.

        """
        return self._browser.tabs


    def get_tab_by_descriptor(self, tab_descriptor):
        """Gets the tab by the tab descriptor.

        @returns: The tab object indicated by the tab descriptor.

        """
        return self._tabs[tab_descriptor]


    @retry_chrome_call
    def close_tab(self, tab_descriptor):
        """Closes the tab.

        @param tab_descriptor: Indicate which tab to be closed.

        """
        if tab_descriptor not in self._tabs:
            raise RuntimeError('There is no tab for %s' % tab_descriptor)
        tab = self._tabs[tab_descriptor]
        del self._tabs[tab_descriptor]
        tab.Close()
        self.clean_unexpected_tabs()


    def wait_for_javascript_expression(
            self, tab_descriptor, expression, timeout):
        """Waits for the given JavaScript expression to be True on the given tab

        @param tab_descriptor: Indicate on which tab to wait for the expression.
        @param expression: Indiate for what expression to wait.
        @param timeout: Indicate the timeout of the expression.
        """
        if tab_descriptor not in self._tabs:
            raise RuntimeError('There is no tab for %s' % tab_descriptor)
        self._tabs[tab_descriptor].WaitForJavaScriptCondition(
                expression, timeout=timeout)


    def execute_javascript(self, tab_descriptor, statement, timeout):
        """Executes a JavaScript statement on the given tab.

        @param tab_descriptor: Indicate on which tab to execute the statement.
        @param statement: Indiate what statement to execute.
        @param timeout: Indicate the timeout of the statement.
        """
        if tab_descriptor not in self._tabs:
            raise RuntimeError('There is no tab for %s' % tab_descriptor)
        self._tabs[tab_descriptor].ExecuteJavaScript(
                statement, timeout=timeout)


    def evaluate_javascript(self, tab_descriptor, expression, timeout):
        """Evaluates a JavaScript expression on the given tab.

        @param tab_descriptor: Indicate on which tab to evaluate the expression.
        @param expression: Indiate what expression to evaluate.
        @param timeout: Indicate the timeout of the expression.
        @return the JSONized result of the given expression
        """
        if tab_descriptor not in self._tabs:
            raise RuntimeError('There is no tab for %s' % tab_descriptor)
        return self._tabs[tab_descriptor].EvaluateJavaScript(
                expression, timeout=timeout)