Source code for cfme.utils.browser

"""Core functionality for starting, restarting, and stopping a selenium browser."""
import atexit
import json
import threading
import time
from collections import namedtuple
from shutil import rmtree
from string import Template
from tempfile import mkdtemp

import os
import requests
import warnings
from cached_property import cached_property
from selenium import webdriver
from selenium.common.exceptions import UnexpectedAlertPresentException, WebDriverException
from selenium.webdriver.common import keys
from selenium.webdriver.common.action_chains import ActionChains
from selenium.webdriver.firefox.firefox_profile import FirefoxProfile
from selenium.webdriver.remote.file_detector import UselessFileDetector
from six.moves.urllib_error import URLError
from werkzeug.local import LocalProxy

from cfme.fixtures.pytest_store import store, write_line
from cfme.utils import conf, tries, clear_property_cache
from cfme.utils.log import logger as log  # TODO remove after artifactor handler
from cfme.utils.path import data_path

# import logging
# log = logging.getLogger('cfme.browser')


BROWSER_ERRORS = URLError, WebDriverException

def _load_firefox_profile():
    # create a firefox profile using the template in data/firefox_profile.js.template

    # Make a new firefox profile dir if it's unset or doesn't exist for some reason
    firefox_profile_tmpdir = mkdtemp(prefix='firefox_profile_')
    log.debug("created firefox profile")
    # Clean up tempdir at exit
    atexit.register(rmtree, firefox_profile_tmpdir, ignore_errors=True)

    template = data_path.join('firefox_profile.js.template').read()
    profile_json = Template(template).substitute(profile_dir=firefox_profile_tmpdir)
    profile_dict = json.loads(profile_json)

    profile = FirefoxProfile(firefox_profile_tmpdir)
    for pref in profile_dict.items():
    return profile

[docs]class Wharf(object): # class level to allow python level atomic removal of instance values docker_id = None def __init__(self, wharf_url): self.wharf_url = wharf_url self._renew_thread = None def _get(self, *args): response = requests.get(os.path.join(self.wharf_url, *args)) if response.status_code == 204: return try: return json.loads(response.content) except ValueError: raise ValueError( "JSON could not be decoded:\n{}".format(response.content))
[docs] def checkout(self): if self.docker_id is not None: return self.docker_id checkout = self._get('checkout') self.docker_id, self.config = next(iter(checkout.items())) self._start_renew_thread()'Checked out webdriver container %s', self.docker_id) log.debug("%r", checkout) return self.docker_id
[docs] def checkin(self): # using dict pop to avoid race conditions my_id = self.__dict__.pop('docker_id', None) if my_id: self._get('checkin', my_id)'Checked in webdriver container %s', my_id) self._renew_thread = None
def _start_renew_thread(self): assert self._renew_thread is None self._renew_thread = threading.Thread(target=self._renew_function) self._renew_thread.daemon = True self._renew_thread.start() def _renew_function(self): # If we have a docker id, renew_timer shouldn't still be None log.debug("renew thread started") while True: time.sleep(FIVE_MINUTES) if self._renew_thread is not threading.current_thread(): log.debug("renew done %s is not %s", self._renew_thread, threading.current_thread()) return if self.docker_id is None: log.debug("renew done, docker id %s", self.docker_id) return expiry_info = self._get('renew', self.docker_id) self.config.update(expiry_info)'Renewed webdriver container %s', self.docker_id) def __nonzero__(self): return self.docker_id is not None
[docs]class BrowserFactory(object): def __init__(self, webdriver_class, browser_kwargs): self.webdriver_class = webdriver_class self.browser_kwargs = browser_kwargs self._add_missing_options() def _add_missing_options(self): if self.webdriver_class is not webdriver.Remote: # desired_capabilities is only for Remote driver, but can sneak in self.browser_kwargs.pop('desired_capabilities', None) elif self.browser_kwargs['desired_capabilities']['browserName'] == 'firefox': self.browser_kwargs['browser_profile'] = self._firefox_profile if self.webdriver_class is webdriver.Firefox: self.browser_kwargs['firefox_profile'] = self._firefox_profile @cached_property def _firefox_profile(self): return _load_firefox_profile()
[docs] def processed_browser_args(self): self._add_missing_options() if 'keep_alive' in self.browser_kwargs: warnings.warn( "forcing browser keep_alive to False due to selenium bugs\n" "we are aware of the performance cost and hope to redeem", category=RuntimeWarning, ) return dict(self.browser_kwargs, keep_alive=False) return self.browser_kwargs
[docs] def create(self, url_key): try: browser = tries( 2, WebDriverException, self.webdriver_class, **self.processed_browser_args()) except URLError as e: if e.reason.errno == 111: # Known issue raise RuntimeError('Could not connect to Selenium server. Is it up and running?') else: # Unknown issue raise browser.file_detector = UselessFileDetector() browser.maximize_window() browser.get(url_key) browser.url_key = url_key return browser
[docs] def close(self, browser): if browser: browser.quit() clear_property_cache(self, '_firefox_profile')
[docs]class WharfFactory(BrowserFactory): def __init__(self, webdriver_class, browser_kwargs, wharf): super(WharfFactory, self).__init__(webdriver_class, browser_kwargs) self.wharf = wharf if browser_kwargs.get('desired_capabilities', {}).get('browserName') == 'chrome': # chrome uses containers to sandbox the browser, and we use containers to # run chrome in wharf, so disable the sandbox if running chrome in wharf co = browser_kwargs['desired_capabilities'].get('chromeOptions', {}) arg = '--no-sandbox' if 'args' not in co: co['args'] = [arg] elif arg not in co['args']: co['args'].append(arg) browser_kwargs['desired_capabilities']['chromeOptions'] = co
[docs] def processed_browser_args(self): command_executor = self.wharf.config['webdriver_url'] view_msg = 'tests can be viewed via vnc on display {}'.format( self.wharf.config['vnc_display'])'webdriver command executor set to %s', command_executor) write_line(view_msg, cyan=True) return dict( super(WharfFactory, self).processed_browser_args(), command_executor=command_executor, )
[docs] def create(self, url_key): def inner(): try: self.wharf.checkout() return super(WharfFactory, self).create(url_key) except URLError as ex: # connection to selenum was refused for unknown reasons log.error('URLError connecting to selenium; recycling container. URLError:') write_line('URLError caused container recycle, see log for details', red=True) log.exception(ex) self.wharf.checkin() raise except Exception: log.exception("failure on webdriver usage, returning container") self.wharf.checkin() raise return tries(WHARF_OUTER_RETRIES, BROWSER_ERRORS, inner)
[docs] def close(self, browser): try: super(WharfFactory, self).close(browser) finally: self.wharf.checkin()
[docs]class BrowserManager(object): def __init__(self, browser_factory): self.factory = browser_factory self.browser = None self._browser_renew_thread = None
[docs] def coerce_url_key(self, key): return key or store.current_appliance.url # TODO: don't rely on store.current_appliance
[docs] def from_conf(cls, browser_conf): webdriver_name = browser_conf.get('webdriver', 'Firefox') webdriver_class = getattr(webdriver, webdriver_name) browser_kwargs = browser_conf.get('webdriver_options', {}) if 'webdriver_wharf' in browser_conf: wharf = Wharf(browser_conf['webdriver_wharf']) atexit.register(wharf.checkin) if browser_conf[ 'webdriver_options'][ 'desired_capabilities']['browserName'].lower() == 'firefox': browser_kwargs['desired_capabilities']['marionette'] = False return cls(WharfFactory(webdriver_class, browser_kwargs, wharf)) else: if webdriver_name == "Remote": if browser_conf[ 'webdriver_options'][ 'desired_capabilities']['browserName'].lower() == 'chrome': browser_kwargs['desired_capabilities']['chromeOptions'] = {} browser_kwargs[ 'desired_capabilities']['chromeOptions']['args'] = ['--no-sandbox'] browser_kwargs['desired_capabilities'].pop('marionette', None) if browser_conf[ 'webdriver_options'][ 'desired_capabilities']['browserName'].lower() == 'firefox': browser_kwargs['desired_capabilities']['marionette'] = False return cls(BrowserFactory(webdriver_class, browser_kwargs))
def _is_alive(self): log.debug("alive check") try: self.browser.current_url except UnexpectedAlertPresentException: # We shouldn't think that an Unexpected alert means the browser is dead return True except Exception: log.exception("browser in unknown state, considering dead") return False return True
[docs] def ensure_open(self, url_key=None): url_key = self.coerce_url_key(url_key) if getattr(self.browser, 'url_key', None) != url_key: return self.start(url_key=url_key) if self._is_alive(): return self.browser else: return self.start(url_key=url_key)
[docs] def add_cleanup(self, callback): assert self.browser is not None try: cl = self.browser.__cleanup except AttributeError: cl = self.browser.__cleanup = [] cl.append(callback)
def _consume_cleanups(self): try: cl = self.browser.__cleanup except AttributeError: pass else: while cl: cl.pop()()
[docs] def quit(self): # TODO: figure if we want to log the url key here self._consume_cleanups() try: self.factory.close(self.browser) except Exception as e: log.error('An exception happened during browser shutdown:') log.exception(e) finally: self.browser = None
[docs] def start(self, url_key=None):'starting browser') url_key = self.coerce_url_key(url_key) if self.browser is not None: self.quit() return self.open_fresh(url_key=url_key)
[docs] def open_fresh(self, url_key=None): url_key = self.coerce_url_key(url_key)'starting browser for %r', url_key) assert self.browser is None self.browser = self.factory.create(url_key=url_key) return self.browser
[docs]class WithZoom(object): """ This class is a decorator that used to wrap function with zoom level. this class perform zoom by <level>, call the target function and exit by zooming back to the original zoom level. Args: * level: int, the zooming value (i.e. -2 -> 2 clicks out; 3 -> 3 clicks in) """ def __init__(self, level): self._level = level def __call__(self, func): def wrapper(*args, **kwargs): ensure_browser_open() with self: return func(*args, **kwargs) return wrapper def __enter__(self, *args, **kwargs): ac = ActionChains(browser()) for _ in range(abs(self._level)): ac.send_keys(keys.Keys.CONTROL, keys.Keys.SUBTRACT if self._level < 0 else keys.Keys.ADD) ac.perform() def __exit__(self, *args, **kwargs): ac = ActionChains(browser()) for _ in range(abs(self._level)): ac.send_keys(keys.Keys.CONTROL, keys.Keys.SUBTRACT if -self._level < 0 else keys.Keys.ADD) ac.perform()
manager = BrowserManager.from_conf(conf.env.get('browser', {})) driver = LocalProxy(manager.ensure_open)
[docs]def browser(): """callable that will always return the current browser instance If ``None``, no browser is running. Returns: The current browser instance. """ return manager.browser
[docs]def ensure_browser_open(url_key=None): """Ensures that there is a browser instance currently open Will reuse an existing browser or start a new one as-needed Returns: The current browser instance. """ if not url_key: from cfme.utils.appliance import current_appliance url_key = current_appliance.server.address() return manager.ensure_open(url_key=url_key)
[docs]def start(url_key=None): """Starts a new web browser If a previous browser was open, it will be closed before starting the new browser Args: """ # Try to clean up an existing browser session if starting a new one return manager.start(url_key=url_key)
[docs]def quit(): """Close the current browser Will silently fail if the current browser can't be closed for any reason. .. note:: If a browser can't be closed, it's usually because it has already been closed elsewhere. """ manager.quit()
ScreenShot = namedtuple("screenshot", ['png', 'error'])
[docs]def take_screenshot(): screenshot = None screenshot_error = None try: screenshot = browser().get_screenshot_as_base64() except (AttributeError, WebDriverException): # See comments utils.browser.ensure_browser_open for why these two exceptions screenshot_error = 'browser error' except Exception as ex: # If this fails for any other reason, # leave out the screenshot but record the reason if str(ex): screenshot_error = '{}: {}'.format(type(ex).__name__, str(ex)) else: screenshot_error = type(ex).__name__ return ScreenShot(screenshot, screenshot_error)