fbpx
Uncategorized

Parallel test execution with Selenium and Python Browser Automation

In a previous blog post, we discussed how to run local tests using Python and Selenium browser automation. In this blog post, we will show you how to configure parallel test execution for a WebRTC product with one or more participants on your local machine.

While being resource-limited, running parallel tests locally has several benefits, mainly faster test script development, and debugging. In addition to the previous blog post, the present aims to demonstrate a test scenario involving two or more participants. The purpose of this illustration is to provide a way of efficiently developing a script for WebRTC test that can later be executed on Loadero with multiple parallel participants. Lastly, the scenario can be simply added to Loadero test via Website GUI or REST API.  However, it’s important to note that a local setup has limitations compared to Loadero, such as resource availability, additional metrics, browser, location support and more. To be able to run the test scenario that we will create in this blog post with thousands of concurrent participants later we recommend starting a free trial of Loadero.

Prerequisites

Before we get started, here are the prerequisites you’ll need:

  • Code editor of your choice (we prefer Visual Studio Code).
  • Python (the latest version is preferable, in this example, v3.11 will be used).
  • Automation environment

WebRTC, an abbreviation for Web Real-Time Communication, is a technology that enables real-time communication functionality within web applications. It is built on an open standard and allows the exchange of video, voice, and messages between peers. The technology is accessible on modern browsers and native clients for leading platforms. Developers can leverage WebRTC to create voice and video communication solutions.

The structure of a WebRTC test can be seen in the chart below.

The underlying technologies of WebRTC are executed as an open web standard and are accessible as typical JavaScript APIs in significant browsers. Further information regarding the WebRTC solution employed in the exemplified test can be found here.

Test scenario: In this test scenario we will use Janus Demo: Video Room App. Two participants simultaneously join the same room and remain in it for two minutes. Throughout the parallel test execution, each participant will verify the presence of the other. If both participants remain connected for the entire two-minute period, they will leave the room, and the test will pass.

Python setup

In a directory called paralel-local-test-execution create a virtual environment with:

python -m venv venv

Then activate the environment with:

source venv/bin/activate

Note: Before running the following command you should get the requirements.txt file. 

Now we have to install all dependencies needed for the test with:

pip install -r requirements.txt

Automation setup

Install browser drivers for Chrome and Firefox. In this blog post we are using MacOS so the installation can be done via brew:

brew install chromedriver geckodriver

After Python and Automation setups are done we can continue with the actual test. Firstly, create a directory called src in the main directory (in this example, parallel-local-test-execution). Once this is done, use the following commands to generate the main test file and helper files:

touch participant.py
touch thread_with_return_value.py
touch test_on_loadero.py

These files will contain your scripts for executing parallel tests.

First, create a class Participant inside participant.py:

class Participant:
    """
    Represents a Loadero participant.

    Each participant is identified by a unique participant ID.
    """

    def __init__(self, participant_id):
        self.participant_id = participant_id

Second, create a class ThreadWithReturnValue inside thread_with_retrun_value.py

from threading import Thread

class ThreadWithReturnValue(Thread):
    """
    A subclass of `Thread` that returns a value when it is joined.

    This class works by overriding the `run` method of the `Thread` class
    to store the return value of the target function in a `_return` attribute.
    When the `join` method is called, it returns the stored return value.

    Args:
        group: The thread group (unused).
        target: The target function to be run in the thread.
        name: The name of the thread.
        args: The positional arguments to be passed to the target function.
        kwargs: The keyword arguments to be passed to the target function.
    """

    def __init__(self, group=None, target=None, name=None,
                    args=(), kwargs=None):
        if kwargs is None:
            kwargs = {}
        Thread.__init__(self, group, target, name, args, kwargs)
        self._return = None

    def run(self):
        if self._target is not None:
            self._return = self._target(*self._args,
                                        **self._kwargs)

    def join(self, *args):
        Thread.join(self, *args)
        return self._return

Last, create the main test file:

  • The initial lines of the code include the importation of essential packages/modules required for the code’s execution. Subsequently, two customized modules, namely NewDriver and TestUIDriver, are imported from the package testui.support. The testui.support package encompasses the fundamental classes required for implementing Selenium WebDriver and TestUIDriver.
import os
import pytest
import sys
import time

from selenium.webdriver.chrome.options import Options
from testui.support.appium_driver import NewDriver
from testui.support.testui_driver import TestUIDriver

from participant import Participant
from thread_with_return_value import ThreadWithReturnValue
  • The program defines a constant called REMOTE_URL, which is initialized with an empty string. Another constant named PARTICIPANTS is defined and assigned an integer value of 1. Additionally, a list named GLOBALS is defined.
REMOTE_URL = ""
PARTICIPANTS = 1
GLOBALS = []
  • The code subsequently verifies the existence of an environment variable titled NUMBER_OF_PARTICIPANTS. If the aforementioned variable exists, the value of PARTICIPANTS is assigned to the integer value of the environment variable. Following this, the int(time.time()) function is leveraged to generate a timestamp, and the code proceeds to execute a loop that iterates an amount of times equal to PARTICIPANTS. During each iteration, a new Participant instance is created with the generated timestamp as a parameter. The timestamp value is incremented by 1 for each subsequent iteration, and the participant_id of each Participant instance is appended to the GLOBALS list. Lastly, the code verifies the existence of an environment variable named SELENIUM_REMOTE_URL. If the aforementioned variable exists, the value of REMOTE_URL is set to the value of the environment variable.
participants_env = os.getenv('NUMBER_OF_PARTICIPANTS')
if participants_env is not None:
    PARTICIPANTS = int(participants_env)

timestamp = int(time.time())
for p in range(PARTICIPANTS):
    participant = Participant(timestamp)
    timestamp += 1
    GLOBALS.append(participant.participant_id)

selenium_url_env = os.getenv('SELENIUM_REMOTE_URL')
if selenium_url_env is not None:
    REMOTE_URL = selenium_url_env
  • A pytest fixture, named driver, has been defined, which is configured to run automatically using the autouse=True parameter. The said fixture creates a list of TestUIDriver instances, with one instance for each participant, as specified by the PARTICIPANTS constant. These TestUIDriver instances are created utilizing the Selenium WebDriver with Chrome and are configured with different options that allow media streams and other settings. In case REMOTE_URL is specified, the drivers will be created with a remote URL instead of a local driver.
# TestUIDriver
@pytest.fixture(autouse=True)
def driver() -> TestUIDriver:
    """Generator function that yields a list of TestUIDriver instances.

    This function creates a list of TestUIDriver instances, one for each participant
    specified by the PARTICIPANTS constant. The TestUIDriver instances are created using
    Selenium WebDriver with Chrome, and are configured with various options to enable
    media streams and other settings. If REMOTE_URL is specified, the drivers will be
    created using a remote URL instead of a local driver.

    Yields:
        A list of TestUIDriver instances.

    Raises:
        ValueError: If PARTICIPANTS is less than 1.
    """
    options = Options()
    chrome_options = ["no-sandbox", "use-fake-device-for-media-stream"]
    prefs = {
        "profile": {
            "content_settings": {
                "exceptions": {
                    "media_stream_camera": {"https://*,*": {"setting": 1}},
                    "media_stream_mic": {"https://*,*": {"setting": 1}},
                }
            },
        }
    }
    options.add_experimental_option("prefs", prefs)
    for o in chrome_options:
        options.add_argument(o)

    drivers = []
    for _ in range(PARTICIPANTS):
        if REMOTE_URL != "":
            driver = (
                NewDriver()
                .set_logger()
                .set_remote_url(REMOTE_URL)
                .set_browser("chrome")
                .set_selenium_driver(chrome_options=options)
            )
        else:
            driver = (
                NewDriver()
                .set_logger()
                .set_browser("chrome")
                .set_selenium_driver(chrome_options=options)
            )

        drivers.append(driver)

    yield drivers

    for participant_driver in drivers:
        participant_driver.quit()
  • A function named test_on_loadero is defined with the pytest.mark.demotest decorator. The function instantiates a list of threads, where each thread is associated with a unique driver in the driver list. Each thread executes the test function with a distinct participant ID and returns the result code. The function raises an exception if any of the threads return a non-zero status code.
# Parallel test execution
@pytest.mark.demotest
def test_on_loadero(driver: TestUIDriver) -> None:
    """Test function that runs Loadero tests for each participant.

    This function creates a list of threads, one for each driver in the driver list.
    Each thread runs the `test` function with a different participant ID, and returns
    the result code. If any of the threads return a non-zero status code, an exception
    is raised.

    Args:
        driver: A list of TestUIDriver instances.
    Raises:
        Exception: If any of the threads return a non-zero status code.
    """
    try:
        threads = []
        participants = []
        for p_id, d in enumerate(driver):
            p = Participant(GLOBALS[p_id])
            participants.append(p)
            t = ThreadWithReturnValue(target=test, args=(d, p.participant_id))
            threads.append(t)
            print('Created thread: ', p.participant_id)
            t.start()
        num_failed = 0
        for t, p in zip(threads, participants):
            status = t.join()
            print(f'Finished thread: {p.participant_id} With code: {status}')
            if status != 0:
                num_failed += 1

        if num_failed > 0:
            raise Exception(f'{num_failed} participants failed')

    except Exception as main_exception:
        print(f'Exception Handled in Main, Details of the Exception: {main_exception}')
        sys.exit(-1)
  • The test function is defined as a method to verify the ability to join and sustain connectivity in a video room with the given WebDriver. It takes two parameters – driver and participant_id. The driver parameter is an instance of the WebDriver that represents the browser utilized for testing, while the participant_id parameter is an integer that uniquely identifies the participant in the test. Upon invocation, the function opens a web page and proceeds to join a video room. After a specified period of time, the function verifies if the participant is still connected to the room. If the connectivity remains uninterrupted, the function returns 0, indicating that the test has passed successfully.
# Test
def test(driver, participant_id):
    """
    Test the ability to join and remain in a video room using the specified WebDriver.

    Parameters:
        driver: A WebDriver instance representing the browser to use for testing.
        participant_id: An integer identifying the participant in the test.

    Returns:
        0 if the test passed successfully, or a non-zero value if an error occurred.
    """
    # room connect options
    url = 'https://janus.conf.meetecho.com/videoroomtest.html'
    identity = 'Participant' + str(participant_id)

    take = 2
    interval = 60
    element_timeout = 35

    # open webpage
    driver.navigate_to(url)
    driver.e('css', 'body').wait_until_visible(element_timeout)  # 35

    # join room
    driver.e('css', '#start').wait_until_visible(element_timeout)  # 35
    driver.e('css', '#start').click()
    driver.e('css', '#username').wait_until_visible(element_timeout)
    driver.e('css', '#username').send_keys(identity)
    driver.e('css', '#register').click()
    time.sleep(interval)

    # check if not Disconnected
    for _ in range(take):
        # page still loaded
        driver.e('css', 'body').wait_until_visible(element_timeout)  # 35s
        # participant not disconnected
        time.sleep(interval)  # 60s
        # disconect
        driver.e('css', '#start').wait_until_visible(element_timeout).click()
    return 0

You can find the script’s code here.

Now your parallel test script is complete and you can start working on your parallel test execution. 

  • Create a file called pytest.ini file that will contain the test marker (in this case demotest).
  • Start parallel test execution script with:
pytest -s ./src/{FILE_NAME}.py -m demotest

To execute your test with parameters, use the following command:

export NUMBER_OF_PARTICIPANTS={NUMBER_OF_PARTICIPANTS}
export SELENIUM_REMOTE_URL={SELENIUM_REMOTE_URL}
pytest -s ./src/{FILENAME}.py -m demotest

This concludes the environment setup and you are ready to develop your Loadero scripts with more than one participant locally. 

Within this blog post we have aimed to inform you on how to run parallel test from your local machine. Once you have composed and validated all the code locally, eliminating any potential errors, you can deploy the test script to Loadero and scale it by executing the test with hundreds or thousands of participants.