Skip to content

Instantly share code, notes, and snippets.

@dmakhno

dmakhno/readme.md

Created February 20, 2014 08:18
Show Gist options
  • Star 1 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save dmakhno/9109015 to your computer and use it in GitHub Desktop.
Save dmakhno/9109015 to your computer and use it in GitHub Desktop.
Download ZIP
Raw
readme.md

You need:

Define module constuructor in init.py

import os
from selenium_utils.page import PageBuilder

from selenium_utils import read_utils_config
from selenium_utils.screenshot import WebScreenShotUtil
from selenium_utils.splinter_wrapper import SplinterDriver, SplinterFactory
from selenium_utils.webdriver import WebDriverManager
#...

__browser_path = os.path.dirname(os.path.realpath(__file__))
CONFIGS = read_utils_config(os.path.join(__browser_path, 'configs/default.yaml'))

class Play2WebDriverManager(WebDriverManager):
    def prepare_hook(self, driver):
        driver.delete_all_cookies()
        if not driver.current_url.startswith(os.getenv('tenant')):
            driver.get(os.getenvn('tenant') + '/404')
        cook = PlatformManage.platform.auth.cookies['PLAY_SESSION'] #this is specific api
        driver.add_cookie({'name': 'PLAY_SESSION', 'value': cook})


WD_MANAGER = Play2WebDriverManager(SplinterFactory(CONFIGS), CONFIGS)
WebScreenShotUtil.screenshot_root = __browser_path
PageBuilder.webdriver_provider = WD_MANAGER #kind a dependency injection

Customize your gui decorator

def with_gui():
    def wrapper(func):

        @wraps(func)
        def wrapped_func(*args, **kwargs):
            self = args[0]
            skip_if_turned_off(self)

            driver = WD_MANAGER.new_driver()
            capture = ScreenshotCapturer(WD_MANAGER, CONFIGS) #almost as wathcer
            session = Site(driver, driver.splinter)

            spec = inspect.getargspec(func)[0]
            for v in ["site", "session", "gui"]:
                if v in spec:
                    kwargs[v] = session

            try:
                func(*args, **kwargs)
            except AssertionError as e:
                capture.on_test_failure(self, None, e)
                #raise for next, to have proper place of issue. (2nd element means traceback in returned tuple
                tb = sys.exc_info()[2].tb_next
                raise e, None, tb
            except Exception as e:
                capture.on_test_error(self, None, e)
                tb = sys.exc_info()[2].tb_next
                raise e, None, tb
            finally:
                do_and_ignore(lambda: WD_MANAGER.close_driver())

        return wrapped_func

    return wrapper
    
def skip_if_turned_off(testcase):
    if str(CONFIGS.get("selenium_active", False)).lower() not in ("yes", "true", "t", "1"):
        name = unicode(type(testcase).__name__) + u"_" + unicode(testcase._testMethodName)
        raise unittest.SkipTest(name + " skipped due to selenium turned off")

Assuming Site facade for gui and construct pages

class Site(object):
    def __init__(self, driver, splinter):
        self.driver = driver
        self.webdriver = driver
        self.splinter = splinter

    def navigate(self, item):
        #translates api item to page
    def to_start(self):
        driver.get(self.tenant)
        return PageBuilder.create_page(StartPage)
    # ... and the same

Note: config file and env variables

# Settings for Selenium WebDriver used for browser testing. Most of the settingf like in wtframework
selenium:
  active: false

  #type: LOCAL
  type: REMOTE

  reusebrowser: true
  threaded: true
  take_screenshot: true
  desired_capabilities: {}
  browser: FIREFOX

Maybe override export wd_selenium_active=true and export wd_selenium_remote_url=http://xxx:4444/wd/hub

Raw
selenium_utils==__init__.py
"""
This module provides common utilities, and have no specific implementations.
Some of the code was borrowed from https://github.com/wiredrive/wtframework
"""
import logging
import os
import yaml
logger = logging.getLogger("selenium_utils")
def read_utils_config(config_file_location):
env_prefix = "wd_"
def read_from_file():
with open(config_file_location, 'r') as config_yaml:
return yaml.load(config_yaml)
def flatten_dict(d, result={}, prv_keys=[]):
for k, v in d.iteritems():
if isinstance(v, dict):
flatten_dict(v, result, prv_keys + [k])
else:
result['_'.join(prv_keys + [k])] = v
return result
configs = read_from_file()
env_config = {k[len(env_prefix):]: v for k, v in os.environ.iteritems() if k.startswith(env_prefix)}
yml_config = flatten_dict(configs)
yml_config.update(env_config) # override values from env
return yml_config
Raw
selenium_utils==page.py
import abc
from functools import wraps
from selenium.common.exceptions import TimeoutException
from selenium.webdriver.support.wait import WebDriverWait
class Page(object):
def __init__(self, webdriver):
self.webdriver = webdriver
__metaclass__ = abc.ABCMeta # needed to make this an abstract class in Python 2.7
@abc.abstractmethod
def validate_page(self):
"""Perform checks to validate this page is the correct target page.
All PageObjects must implement this method.
Args:
webdriver (Webdriver) : instance of Selenium Webdriver
Raises:
InvalidPageError: Raised when we try to assign the wrong page
to this page object. This exception should be raised when a page match
fails. Any other exception type would be consider a code failure.
Returns:
self: for fluidity
"""
return
def on_page(self):
try:
self.validate_page()
return True
except InvalidPageError:
return False
def _wait_page_load(self, method, message):
try:
WebDriverWait(self.webdriver, 10).until(method)
except TimeoutException:
raise InvalidPageError(message)
@staticmethod
def validate():
"""
Decorator, for page methods, that validate that page is still open
"""
def wrapper(func):
@wraps(func)
def wrapped_func(*args, **kwargs):
self = args[0]
assert isinstance(self, Page)
assert self.on_page()
return func(*args, **kwargs)
return wrapped_func
return wrapper
class PageBuilder(object):
webdriver_provider = None
@staticmethod
def create_page(page_obj_class, **kwargs):
webdriver = PageBuilder.webdriver_provider.get_driver()
page = page_obj_class(webdriver, **kwargs)
class InvalidPageError(Exception):
'''Thrown when we have tried to instantiate the incorrect page to a PageObject.'''
pass
Raw
selenium_utils==screenshot.py
"""
95% of code from wtframework, hope do not break license
"""
import datetime
import re
import base64
import os
from selenium.webdriver import remote
class ScreenshotCapturer(object):
@staticmethod
def generate_screenshot_filename(testcase):
'''
Get the class name and timestamp for generating filenames
Return:
str - File Name.
'''
fname = unicode(type(testcase).__name__) + u"_" + unicode(testcase._testMethodName)
fname = re.sub(u"[^a-zA-Z_]+", u"_", fname)
# Trim test case name incase it's too long.
fname = fname[:100]
fmt = u'%y-%m-%d_%H.%M.%S_{fname}'
return datetime.datetime.now().strftime(fmt).format(fname=fname)
def __init__(self, webdriver_provider, config):
self.capture_screenshot = config.get("selenium_take_screenshot", True)
self.annotate_screenshot = config.get("selenium_annotate_screenshot", False)
self._screenshot_util = WebScreenShotUtil
self._webdriver_provider = webdriver_provider
def on_test_failure(self, testcase, test_result, exception):
"""
On test failure capture screenshot handler.
"""
name = self.generate_screenshot_filename(testcase) + u"_Fail"
print "FAIL: " + exception.message
if self.capture_screenshot:
self.__take_screenshot_if_webdriver_open__(name)
if self.annotate_screenshot: self.__annotate_screenshot__(name, exception)
def on_test_error(self, testcase, test_result, exception):
"""
On test error, capture screenshot handler.
"""
name = self.generate_screenshot_filename(testcase) + u"_Error"
print "ERROR: " + exception.message
if self.capture_screenshot:
self.__take_screenshot_if_webdriver_open__(name)
if self.annotate_screenshot: self.__annotate_screenshot__(name, exception)
def __annotate_screenshot__(self, name, exception):
"""
This method puts short error information on screenshot
"""
assert False, "not implemented"
# from wtframework.wtf.utils.project_utils import ProjectUtils
# import os
# file_location = os.path.join(ProjectUtils.get_project_root(),
# WebScreenShotUtil.SCREEN_SHOT_LOCATION,
# name + ".png")
#
# from PIL import Image
# from PIL import ImageFont
# from PIL import ImageDraw
# img = Image.open(file_location)
# draw = ImageDraw.Draw(img)
# font = ImageFont.truetype("sans-serif.ttf", 16)
# draw.text((0, 0),exception.message,(255,255,255),font=font)
# img.save(file_location)
def __take_screenshot_if_webdriver_open__(self, name):
'''
Take a screenshot if webdriver is open.
Args:
testcase: TestCase
'''
if self._webdriver_provider.is_driver_available():
try:
self._screenshot_util.take_screenshot(self._webdriver_provider.get_driver(), name)
print u"Screenshot taken: " + name
except Exception as e:
print u"Unable to take screenshot. Reason: " + e.message + unicode(type(e))
class WebScreenShotUtil():
'''
Utilities for taking screenshots in Selenium Webdriver.
'''
screenshot_root = ""
SCREEN_SHOT_LOCATION = "screenshots"
REFERENCE_SCREEN_SHOT_LOCATION = "reference-screenshots"
@staticmethod
def take_screenshot(webdriver, file_name):
"""
Captures a screenshot.
Args:
webdriver (WebDriver) - Selenium webdriver.
file_name (str) - File name to save screenshot as.
"""
folder_location = os.path.join(WebScreenShotUtil.screenshot_root,
WebScreenShotUtil.SCREEN_SHOT_LOCATION)
WebScreenShotUtil.__capture_screenshot(webdriver, folder_location, file_name + ".png")
@staticmethod
def take_reference_screenshot(webdriver, file_name):
"""
Captures a screenshot as a reference screenshot.
Args:
webdriver (WebDriver) - Selenium webdriver.
file_name (str) - File name to save screenshot as.
"""
folder_location = os.path.join(WebScreenShotUtil.screenshot_root,
WebScreenShotUtil.REFERENCE_SCREEN_SHOT_LOCATION)
WebScreenShotUtil.__capture_screenshot(webdriver, folder_location, file_name + ".png")
@staticmethod
def __capture_screenshot(webdriver, folder_location, file_name):
"Capture a screenshot"
# Check folder location exists.
if not os.path.exists(folder_location): os.makedirs(folder_location)
file_location = os.path.join(folder_location, file_name)
if isinstance(webdriver, remote.webdriver.WebDriver):
# If this is a remote webdriver. We need to transmit the image data
# back across system boundries as a base 64 encoded string so it can
# be decoded back on the local system and written to disk.
base64_data = webdriver.get_screenshot_as_base64()
screenshot_data = base64.decodestring(base64_data)
screenshot_file = open(file_location, "wb")
screenshot_file.write(screenshot_data)
screenshot_file.close()
else:
webdriver.save_screenshot(file_location)
Raw
selenium_utils==splinter_wrapper.py
from selenium.webdriver.support.wait import WebDriverWait
from splinter.cookie_manager import CookieManagerAPI
from splinter.driver.webdriver import BaseWebDriver
from splinter.driver.webdriver.remote import WebDriverElement as RemoteWebDriverElement
from webdriver import WebDriverFactory
class SplinterDriver(BaseWebDriver):
"""
Wraps existing webdriver and allows to use splinter api
"""
driver_name = "Remote webdriver wrapper"
def __init__(self, existing_driver, wait_time=2):
self.driver = existing_driver
self.element_class = ExWebDriverElement
self._cookie_manager = CookieManagerAPI()
super(SplinterDriver, self).__init__(wait_time)
def wait(self, method, wait_time=2):
return WebDriverWait(self.driver, wait_time).until(method)
class ExWebDriverElement(RemoteWebDriverElement):
def as_select(self):
"""
Converts WebElement to Select object, not splinter part, because WebDriver specific
"""
from selenium.webdriver.support.ui import Select
return Select(self._element)
@property
def selected_text(self):
return self.as_select().first_selected_option.text
class SplinterFactory(WebDriverFactory):
"""
Injects splinter driver into instance of webdriver.
Hacky, however simplifies conversions
"""
def post_create(self, driver):
super(SplinterFactory, self).post_create(driver)
driver.splinter = SplinterDriver(driver)
Raw
selenium_utlis==webdriver.py
"""
95% of code from wtframework, hope do not break license
"""
import os
from threading import current_thread
import time
import urllib2
from selenium import webdriver
from selenium.webdriver.common.desired_capabilities import DesiredCapabilities
from . import logger
class WebDriverFactory(object):
'''
This class constructs a Selenium Webdriver using settings in the config file.
This allows you to substitute different webdrivers by changing the config settings
while keeping your tests using the same webdriver interface.
Ideally you will not use this directly. You will normally use WTF_WEBDRIVER_MANAGER.new_driver()
to create a new instance of webdriver.
You can extend this class for the purposes of adding support for webdrivers that are not
currently supported.
'''
# Note: please be sure to uncomment the Unit test and run them manually before
# pushing any changes. This is because they are disabled. The reason is
# because the unit tests for this class can use up billable hours on sauce labs
# or open annoying browser windows.
# CONFIG SETTINGS #
DRIVER_TYPE_CONFIG = "selenium_type"
REMOTE_URL_CONFIG = "selenium_remote_url"
BROWSER_TYPE_CONFIG = "selenium_browser"
DESIRED_CAPABILITIES_CONFIG = "selenium_desired_capabilities"
CHROME_DRIVER_PATH = "selenium_chromedriver_path"
PHANTOMEJS_EXEC_PATH = "selenium_phantomjs_path"
SELENIUM_SERVER_LOCATION = "selenium_selenium_server_path"
LOG_REMOTEDRIVER_PROPS = "selenium_log_remote_webdriver_props"
# BROWSER CONSTANTS #
HTMLUNIT = "HTMLUNIT"
HTMLUNITWITHJS = "HTMLUNITWITHJS"
ANDROID = "ANDROID"
CHROME = "CHROME"
FIREFOX = "FIREFOX"
INTERNETEXPLORER = "INTERNETEXPLORER"
IPAD = "IPAD"
IPHONE = "IPHONE"
OPERA = "OPERA"
PHANTOMJS = "PHANTOMJS"
SAFARI = "SAFARI"
# ENV vars that are used by selenium.
__SELENIUM_SERVER_JAR_ENV = "SELENIUM_SERVER_JAR"
def __init__(self, config):
'''
Initializer.
'''
self._config = config
def create_webdriver(self, testname=None):
'''
Creates an instance of Selenium webdriver based on config settings.
This should only be called by a shutdown hook. Do not call directly within
a test.
Kwargs:
testname: Optional test name to pass, this gets appended to the test name
sent to selenium grid.
Returns:
WebDriver - Selenium webdriver instance.
'''
try:
driver_type = self._config.get(WebDriverFactory.DRIVER_TYPE_CONFIG)
except:
print WebDriverFactory.DRIVER_TYPE_CONFIG + " setting is missing from config. Using defaults"
driver_type = "LOCAL"
if driver_type == "REMOTE":
# Create desired capabilities.
driver = self.__create_remote_webdriver_from_config(testname=testname)
else:
# handle as local webdriver
driver = self.__create_driver_from_browser_config()
self.post_create(driver)
return driver
def post_create(self, driver):
try:
driver.maximize_window()
except:
# wait a short period and try again.
time.sleep(5000)
try:
driver.maximize_window()
except Exception as e:
logger.warn(
"Unable to maximize browser window. It may be possible the browser did not instantiate correctly.",
e)
def __create_driver_from_browser_config(self):
'''
Reads the config value for browser type.
'''
try:
browser_type = self._config.get(WebDriverFactory.BROWSER_TYPE_CONFIG)
except KeyError:
print WebDriverFactory.BROWSER_TYPE_CONFIG + " missing is missing from config file. Using defaults"
browser_type = WebDriverFactory.FIREFOX
browser_type_dict = {
WebDriverFactory.CHROME: lambda: webdriver.Chrome(
self._config.get(WebDriverFactory.CHROME_DRIVER_PATH)),
WebDriverFactory.FIREFOX: lambda: webdriver.Firefox(),
WebDriverFactory.INTERNETEXPLORER: lambda: webdriver.Ie(),
WebDriverFactory.OPERA: lambda: webdriver.Opera(),
WebDriverFactory.PHANTOMJS: lambda: self.__create_phantom_js_driver(),
WebDriverFactory.SAFARI: lambda: self.__create_safari_driver()
}
try:
return browser_type_dict[browser_type]()
except KeyError:
raise TypeError("Unsupported Browser Type {0}".format(browser_type))
def __create_safari_driver(self):
'''
Creates an instance of Safari webdriver.
'''
# Check for selenium jar env file needed for safari driver.
if not os.getenv(self.__SELENIUM_SERVER_JAR_ENV):
# If not set, check if we have a config setting for it.
try:
selenium_server_path = self._config.get(self.SELENIUM_SERVER_LOCATION)
os.environ[self.__SELENIUM_SERVER_JAR_ENV] = selenium_server_path
except KeyError:
raise RuntimeError("Missing selenium server path config {0}.".format(self.SELENIUM_SERVER_LOCATION))
return webdriver.Safari()
def __create_phantom_js_driver(self):
'''
Creates an instance of PhantomJS driver.
'''
try:
return webdriver.PhantomJS(executable_path=self._config.get(self.PHANTOMEJS_EXEC_PATH),
service_args=['--ignore-ssl-errors=true'])
except KeyError:
return webdriver.PhantomJS(service_args=['--ignore-ssl-errors=true'])
def __create_remote_webdriver_from_config(self, testname=None):
'''
Reads the config value for browser type.
'''
browser_type = self._config.get(WebDriverFactory.BROWSER_TYPE_CONFIG)
remote_url = self._config.get(WebDriverFactory.REMOTE_URL_CONFIG)
browser_constant_dict = {WebDriverFactory.HTMLUNIT: DesiredCapabilities.HTMLUNIT,
WebDriverFactory.HTMLUNITWITHJS: DesiredCapabilities.HTMLUNITWITHJS,
WebDriverFactory.ANDROID: DesiredCapabilities.ANDROID,
WebDriverFactory.CHROME: DesiredCapabilities.CHROME,
WebDriverFactory.FIREFOX: DesiredCapabilities.FIREFOX,
WebDriverFactory.INTERNETEXPLORER: DesiredCapabilities.INTERNETEXPLORER,
WebDriverFactory.IPAD: DesiredCapabilities.IPAD,
WebDriverFactory.IPHONE: DesiredCapabilities.IPHONE,
WebDriverFactory.OPERA: DesiredCapabilities.OPERA,
WebDriverFactory.SAFARI: DesiredCapabilities.SAFARI,
WebDriverFactory.PHANTOMJS: DesiredCapabilities.PHANTOMJS}
try:
desired_capabilities = browser_constant_dict[browser_type].copy()
except KeyError:
raise TypeError("Unsupported Browser Type {0}".format(browser_type))
# Get additional desired properties from config file and add them in.
other_desired_capabilities = self._config.get(WebDriverFactory.DESIRED_CAPABILITIES_CONFIG, {})
for prop in other_desired_capabilities:
value = other_desired_capabilities[prop]
if type(other_desired_capabilities[prop]) is dict:
# do some recursive call to flatten this setting.
self.__flatten_capabilities(desired_capabilities, prop, other_desired_capabilities[prop])
else: # Handle has a single string value.
if isinstance(value, basestring):
desired_capabilities[prop] = value
elif prop == "version": # Version is specified as a string, but we'll allow user to use an int for convenience.
desired_capabilities[prop] = str(value)
else:
desired_capabilities[prop] = value
# Set the test name property if specified in the WTF_TESTNAME var.
try:
test_name = self._config.get("TESTNAME")
desired_capabilities['name'] = test_name
except KeyError:
pass # No test name is specified, use the default.
# Append optional testname postfix if supplied.
if testname:
if desired_capabilities['name']:
desired_capabilities['name'] += "-" + testname
else:
# handle case where name is not specified.
desired_capabilities['name'] = testname
# Instantiate remote webdriver.
driver = webdriver.Remote(
desired_capabilities=desired_capabilities,
command_executor=remote_url
)
# Log IP Address of node if configured, so it can be used to troubleshoot issues if they occur.
log_driver_props = str(self._config.get(WebDriverFactory.LOG_REMOTEDRIVER_PROPS, False)).lower() in ["true",
"yes"]
if "wd/hub" in remote_url and log_driver_props:
try:
grid_addr = remote_url[:remote_url.index("wd/hub")]
info_request_response = urllib2.urlopen(grid_addr + "grid/api/testsession?session=" + driver.session_id,
"", 5000)
node_info = info_request_response.read()
logger.info("RemoteWebdriver using node: " + str(node_info).strip())
except:
# Unable to get IP Address of remote webdriver.
# This happens with many 3rd party grid providers as they don't want you accessing info on nodes on
# their internal network.
pass
return driver
# End of method.
def __flatten_capabilities(self, desired_capabilities, prefix, setting_group):
for key in setting_group.keys():
if type(setting_group[key]) is dict:
# Do recursive call
self.__flatten_capabilities(desired_capabilities, prefix + "." + key, setting_group[key])
else:
value = setting_group[key]
if isinstance(value, basestring):
desired_capabilities[prefix + "." + key] = value
else:
desired_capabilities[prefix + "." + key] = str(value)
class WebDriverManager(object):
'''
Provides Singleton instance of Selenium WebDriver based on
config settings.
Reason we don't make this a Utility class that provides a singleton
of the WebDriver itself is so we can allow that piece to be mocked
out to assist in unit testing framework classes that may use this.
'''
"Config setting to reuse browser instances between WebdriverManager.new_driver() calls."
REUSE_BROWSER = "selenium_reusebrowser"
"Config setting to automatically tear down webdriver upon exiting the main script."
SHUTDOWN_HOOK_CONFIG = "selenium_shutdown_hook"
"Config setting to use new webdriver instance per thread."
ENABLE_THREADING_SUPPORT = "selenium_threaded"
def __init__(self, webdriver_factory, config):
'''
Initializer
Kwargs:
webdriver_factory (WebDriverFactory): Override default webdriver factory.
config (ConfigReader): Override default config reader.
'''
self.__webdriver = {} # Object with channel as a key
self.__registered_drivers = {}
self.__use_shutdown_hook = config.get(WebDriverManager.SHUTDOWN_HOOK_CONFIG, True)
self.__reuse_browser = config.get(WebDriverManager.REUSE_BROWSER, True)
self.__enable_threading_support = config.get(WebDriverManager.ENABLE_THREADING_SUPPORT, False)
self._webdriver_factory = webdriver_factory
def prepare_hook(self, driver):
"""
Attempt to get the browser to a pristine state as possible when we are
reusing this for another test.
If reuse browser is set, we'll avoid closing it and just clear out the cookies,
and reset the location.
Hook allows to override this behaviour
"""
driver.delete_all_cookies()
driver.get("about:blank") # check to see if webdriver is still responding
def clean_up_webdrivers(self):
'''
Clean up webdrivers created during execution.
'''
try:
if self.__use_shutdown_hook:
for key in self.__registered_drivers.keys():
for driver in self.__registered_drivers[key]:
try:
driver.quit()
except:
pass
except:
pass
def close_driver(self):
"""
Close current instance of webdriver.
"""
channel = self.__get_channel()
driver = self.__get_driver_for_channel(channel)
if not self.__reuse_browser and driver:
try:
driver.quit()
except:
pass
self.__unregister_driver(channel)
if driver in self.__registered_drivers[channel]:
self.__registered_drivers[channel].remove(driver)
def get_driver(self):
'''
Get an already running instance of webdriver. If there is none, it will create one.
Returns:
WebDriver - Selenium Webdriver instance.
'''
return self.__get_driver_for_channel(self.__get_channel()) or self.new_driver()
def is_driver_available(self):
'''
Check if a webdriver instance is created.
Returns:
bool - True, webdriver is available; False, webdriver not yet initialized.
'''
channel = self.__get_channel()
try:
return self.__webdriver[channel] != None
except:
return False
def new_driver(self, testname=None):
'''
Used at a start of a test to get a new instance of webdriver. If the
'resuebrowser' setting is true, it will use a recycled webdriver instance.
Kwargs:
testname (str) - Optional test name to pass to Selenium Grid.
Returns:
Webdriver - Selenium Webdriver instance.
'''
channel = self.__get_channel()
# Get reference for the current driver.
driver = self.__get_driver_for_channel(channel)
if self.__reuse_browser:
if not driver:
driver = self._webdriver_factory.create_webdriver(testname=testname)
self.prepare_hook(driver)
# Register webdriver so it can be retrieved by the manager and cleaned up after exit.
self.__register_driver(channel, driver)
else:
try:
self.prepare_hook(driver)
except:
# In the case the browser is unhealthy, we should kill it and serve a new one.
try:
if driver.is_online():
logger.debug("Closing webdriver for thread due to unhealthy: ", channel)
driver.quit()
except:
pass
driver = self._webdriver_factory.create_webdriver(testname=testname)
self.__register_driver(channel, driver)
else:
# Attempt to tear down any existing webdriver.
if driver:
try:
logger.debug("Closing webdriver for thread: ", channel)
driver.quit()
except:
pass
self.__unregister_driver(channel)
driver = self._webdriver_factory.create_webdriver(testname=testname)
self.prepare_hook(driver)
self.__register_driver(channel, driver)
return driver
# End of new_driver method.
def __register_driver(self, channel, webdriver):
"Register webdriver to a channel."
# Add to list of webdrivers to cleanup.
if not self.__registered_drivers.has_key(channel):
self.__registered_drivers[channel] = [] # set to new empty array
self.__registered_drivers[channel].append(webdriver)
# Set singleton instance for the channel
self.__webdriver[channel] = webdriver
def __unregister_driver(self, channel):
"Unregister webdriver"
driver = self.__get_driver_for_channel(channel)
if self.__registered_drivers.has_key(channel) and driver in self.__registered_drivers[channel]:
self.__registered_drivers[channel].remove(driver)
self.__webdriver[channel] = None
def __get_driver_for_channel(self, channel):
"Get webdriver for channel"
try:
return self.__webdriver[channel]
except:
return None
def __get_channel(self):
"Get the channel to register webdriver to."
return current_thread().ident if self.__enable_threading_support else 0
def __del__(self):
"Deconstructor, call cleanup drivers."
self.clean_up_webdrivers()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment