140 lines
4.3 KiB
Python
140 lines
4.3 KiB
Python
# 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
|
|
|
|
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
|
|
|
|
_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 FacadeResource(object):
|
|
"""This class provides access to telemetry chrome wrapper."""
|
|
|
|
EXTRA_BROWSER_ARGS = ['--enable-gpu-benchmarking']
|
|
|
|
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.
|
|
|
|
"""
|
|
if chrome_object:
|
|
self._chrome = chrome_object
|
|
else:
|
|
self._chrome = chrome.Chrome(
|
|
extension_paths=[constants.MULTIMEDIA_TEST_EXTENSION],
|
|
extra_browser_args=self.EXTRA_BROWSER_ARGS,
|
|
clear_enterprise_policy=not restart,
|
|
autotest_ext=True)
|
|
self._browser = self._chrome.browser
|
|
# 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()
|
|
|
|
|
|
def close(self):
|
|
"""Closes Chrome."""
|
|
self._chrome.close()
|
|
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
|
|
def __exit__(self, *args):
|
|
self.close()
|
|
|
|
|
|
@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, 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
|
|
|
|
|
|
@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_descriptor = hex(id(tab))
|
|
self._tabs[tab_descriptor] = tab
|
|
return tab_descriptor
|
|
|
|
|
|
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()
|