PART 1: creating wrapper for the browser management

BUILDING TEST AUTOMATION FRAMEWORK WITH TEST JUNKIE & SELENIUM WEBDRIVER

THE GOAL

Create a wrapper to abstract WebDriver API, generalize browser management functionality and support multi-threading.

THE HOW

As with any new project, I'm going to start by defining the project structure. If you choose to follow this tutorial, I recommend to take a look at the provided project structure and replicate it in your project, before you go any further. We will, also, need to install some packages from PyPi:

  1. test-junkie: This entire tutorial series would not be possible without this package as it provides much of the functionality that we will need. Such as parametrization, parallel test execution and extensive run configurations. Install it with pip install test-junkie or python -m pip install test-junkie
  2. webdriver-manager: This package will simplify our life. It will install and return the correct path to any of the Drivers that we may need for testing. If you are not familiar with this package I recommend to watch this video where I show how to use this package and why it it so useful. Install it with pip install webdriver-manager or python -m pip install webdriver-manager

project structure

Now lets write some code for the browser wrapper. I'm going to provide the code snippet bellow for the wrapper object. Feel free to copy-paste the code into the Browser.py file.

from selenium import webdriver
from webdriver_manager.chrome import ChromeDriverManager


class Browser:

    @staticmethod
    def __chrome(**kwargs):

        options = webdriver.ChromeOptions()

        if kwargs.get("proxy_port"):
            options.add_argument('--proxy-server={proxy}'.format(proxy=kwargs.get("proxy_port")))
        if kwargs.get("headless"):
            options.add_argument('--headless')
        if kwargs.get("no_sandbox"):
            options.add_argument('--no-sandbox')
        if kwargs.get("disable_shm"):
            options.add_argument('--disable-dev-shm-usage')
        if kwargs.get("disable_notifications"):
            options.add_argument('--disable-notifications')
        if kwargs.get("window_size"):
            options.add_argument('--window-size={}'.format(kwargs.get("window_size")))
        if kwargs.get("dev_tools_port"):
            options.add_argument('--remote-debugging-port={port}'.format(port=kwargs.get("dev_tools_port")))
        # any other options you want to support

        driver = webdriver.Chrome(ChromeDriverManager().install(), chrome_options=options)
        driver.set_page_load_timeout(kwargs.get("page_load_timeout", 20))
        return driver

    @staticmethod
    def __ie(**kwargs):
        # If you test on IE, I'm sorry!
        pass

    @staticmethod
    def __ff(**kwargs):
        # If you test on FireFox, add your code here similar to Browser.__chrome.
        pass

    @staticmethod
    def get_driver(**kwargs):
        return Yolo.get_target(target=getattr(Browser, kwargs.get("target", "_Browser__chrome")),
                               freak_mode=kwargs.get("freak_mode", False),
                               **kwargs)

    @staticmethod
    def shutdown(**kwargs):
        Browser.get_driver().quit()
        Yolo.remove_target(target=getattr(Browser, kwargs.get("target", "_Browser__chrome")),
                           freak_mode=kwargs.get("freak_mode", False))

    @staticmethod
    def back():
        Browser.get_driver().back()

    @staticmethod
    def forward():
        Browser.get_driver().forward()
    # All functions that control the browser such as open/close tabs, navigate to a page etc should go into this class

Browser class is going to be the only interface from now on to get the Driver instance for any of the browsers. In this tutorial I'm only focusing on the ChromeDriver, however.

Yolo (for lack of a better term) is a utility class that I created to help me create thread aware context functions. I'm providing a code snippet for that as well, which you can place in the same Browser.py file right after the Browser class.
class Yolo:

    __MAP = {}

    @staticmethod
    def __get_caller():
        import threading
        return threading.current_thread()

    @staticmethod
    def get_target(target, **kwargs):
        """
        When you call Yolo.get_target() initially, the object created from the target argument gets mapped to the
        thread and then returned, any subsequent calls to Yolo
        to get the same object from the same thread will return the exact same instance of the object. However, if the
        call is made from a different thread, the object will be created again for that specific thread.
        Yolo was created to solve a primary use case for creating Page Objects so that you never have to manage and
        pass the driver instance from Page Object to Page Object. It means that you can just ask for a driver from
        any of the Page Object and be sure that a valid instance of the driver will be returned to you even when
        you are creating many instances of the same page in any of your tests and running them in parallel.
        :param target: Runnable FUNCTION/METHOD. Must returns an instance of an object when executed.
        :param kwargs: Any KWARGS that you want to pass in to your :param target: or any other properties
                       you want to support
        :return: Object created by the :param target:()
        """
        caller_thread = Yolo.__get_caller()
        if caller_thread not in Yolo.__MAP:
            Yolo.__MAP.update({caller_thread: {target: target(**kwargs)}})
        elif target not in Yolo.__MAP[caller_thread]:
            Yolo.__MAP[caller_thread].update({target: target(**kwargs)})
        return Yolo.__MAP[caller_thread][target]

    @staticmethod
    def remove_target(target):
        """
        Remove any previously mapped target
        :param target: Runnable FUNCTION/METHOD. Must returns an instance of an object when executed.
        :return: BOOLEAN, True/False depending on if the target was removed
        """
        caller_thread = Yolo.__get_caller()
        if caller_thread in Yolo.__MAP and target in Yolo.__MAP[caller_thread]:
            Yolo.__MAP[caller_thread].pop(target)
            return True
        return False
Anytime a call is made to the Browser.get_driver(), Yolo will check which thread made the call and it will also check if the target object was already created for that thread, if it was, then it will return already created object. However, if object was not created yet, it will run the target function which must return a new object and in our case it returns a ChromeDriver instance which then will be mapped and returned for all of the consecutive calls made by the same thread. So there it is - first abstraction layer and with just over 120 lines of code, our framework has multi-threading support at it's core!

Its worth to note that once Yolo maps an object, there is no way to get a new one, without removing the old one first. That's exactly what we are doing in Browser.quit(). So if we follow the quit call with another Browser.get_driver() call, a completely new driver will be created and its perfectly fine because the old one was closed and can no longer be operated on, but we did request a driver so Yolo will return a new one.

TEST IT

Lets make sure the code works. I want to verify that the code can run in parallel and keep correct context per each browser instance. To do this, I will use the code bellow which will create three threads and each of those threads will have to run 3 different queries on Google that are themed.

import threading
import time
from random import randint
from your.project import Browser

def search(query_strings):
    driver = Browser.get_driver()
    for query in query_strings:
        driver.get("https://www.google.com/search?q={}".format(query))
        time.sleep(randint(1, 5))  # just to make it a bit more interesting

queries = [["test+junkie+selenium", "test+junkie+webdriver", "test+junkie+testing+framework"],
           ["cats", "cute+cats", "tigers"],
           ["dogs", "cute+dogs", "wolfs"]]

threads = []
for query_set in queries:
    thread = threading.Thread(target=search, args=(query_set,))
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

Note that I did not assign a driver to any of the threads, I'm just calling the standard method Browser.get_driver(). When queries run, I expect the same browser to run queries that relate to one theme and one theme only. So if a browser started by running the cats query, it should not run any other queries for dogs or Test Junkie. I did not shutdown the browsers so that I can go and check the query history in each of the browsers. Next >>

footer-background-top