672 lines
		
	
	
		
			25 KiB
		
	
	
	
		
			Python
		
	
	
	
	
	
			
		
		
	
	
			672 lines
		
	
	
		
			25 KiB
		
	
	
	
		
			Python
		
	
	
	
	
	
# pylint: disable=missing-docstring
 | 
						|
import os
 | 
						|
import sys
 | 
						|
import time
 | 
						|
import logging
 | 
						|
from typing import Union, Optional
 | 
						|
import warnings
 | 
						|
import percy
 | 
						|
import requests
 | 
						|
 | 
						|
from selenium import webdriver
 | 
						|
from selenium.webdriver.support import expected_conditions as EC
 | 
						|
from selenium.webdriver.common.by import By
 | 
						|
from selenium.webdriver.support.wait import WebDriverWait
 | 
						|
from selenium.webdriver.common.keys import Keys
 | 
						|
from selenium.webdriver.common.action_chains import ActionChains
 | 
						|
 | 
						|
from selenium.common.exceptions import (
 | 
						|
    WebDriverException,
 | 
						|
    TimeoutException,
 | 
						|
    MoveTargetOutOfBoundsException,
 | 
						|
)
 | 
						|
 | 
						|
from dash.testing.wait import (
 | 
						|
    text_to_equal,
 | 
						|
    style_to_equal,
 | 
						|
    class_to_equal,
 | 
						|
    contains_text,
 | 
						|
    contains_class,
 | 
						|
    until,
 | 
						|
)
 | 
						|
from dash.testing.dash_page import DashPageMixin
 | 
						|
from dash.testing.errors import DashAppLoadingError, BrowserError, TestingTimeoutError
 | 
						|
from dash.testing.consts import SELENIUM_GRID_DEFAULT
 | 
						|
 | 
						|
 | 
						|
logger = logging.getLogger(__name__)
 | 
						|
 | 
						|
 | 
						|
class Browser(DashPageMixin):
 | 
						|
    _url: str
 | 
						|
 | 
						|
    # pylint: disable=too-many-arguments
 | 
						|
    def __init__(
 | 
						|
        self,
 | 
						|
        browser: str,
 | 
						|
        remote: bool = False,
 | 
						|
        remote_url: Optional[str] = None,
 | 
						|
        headless: bool = False,
 | 
						|
        options: Optional[Union[dict, list]] = None,
 | 
						|
        download_path: str = "",
 | 
						|
        percy_run: bool = True,
 | 
						|
        percy_finalize: bool = True,
 | 
						|
        percy_assets_root: str = "",
 | 
						|
        wait_timeout: int = 10,
 | 
						|
        pause: bool = False,
 | 
						|
    ):
 | 
						|
        self._browser = browser.lower()
 | 
						|
        self._remote_url = remote_url
 | 
						|
        self._remote = (
 | 
						|
            True if remote_url and remote_url != SELENIUM_GRID_DEFAULT else remote
 | 
						|
        )
 | 
						|
        self._headless = headless
 | 
						|
        self._options = options
 | 
						|
        self._download_path = download_path
 | 
						|
        self._wait_timeout = wait_timeout
 | 
						|
        self._percy_finalize = percy_finalize
 | 
						|
        self._percy_run = percy_run
 | 
						|
        self._pause = pause
 | 
						|
 | 
						|
        self._driver = until(self.get_webdriver, timeout=1)
 | 
						|
        self._driver.implicitly_wait(2)
 | 
						|
 | 
						|
        self._wd_wait = WebDriverWait(self.driver, wait_timeout)
 | 
						|
        self._last_ts = 0
 | 
						|
        self._url = ""
 | 
						|
 | 
						|
        self._window_idx = 0  # switch browser tabs
 | 
						|
 | 
						|
        if self._percy_run:
 | 
						|
            self.percy_runner = percy.Runner(
 | 
						|
                loader=percy.ResourceLoader(
 | 
						|
                    webdriver=self.driver,
 | 
						|
                    base_url="/assets",
 | 
						|
                    root_dir=percy_assets_root,
 | 
						|
                )
 | 
						|
            )
 | 
						|
            self.percy_runner.initialize_build()
 | 
						|
 | 
						|
        logger.debug("initialize browser with arguments")
 | 
						|
        logger.debug("  headless => %s", self._headless)
 | 
						|
        logger.debug("  download_path => %s", self._download_path)
 | 
						|
        logger.debug("  percy asset root => %s", os.path.abspath(percy_assets_root))
 | 
						|
 | 
						|
    def __enter__(self):
 | 
						|
        return self
 | 
						|
 | 
						|
    def __exit__(self, exc_type, exc_val, traceback):
 | 
						|
        try:
 | 
						|
            self.driver.quit()
 | 
						|
            if self._percy_run and self._percy_finalize:
 | 
						|
                logger.info("percy runner finalize build now")
 | 
						|
                self.percy_runner.finalize_build()
 | 
						|
            else:
 | 
						|
                logger.info("percy finalize relies on CI job")
 | 
						|
        except WebDriverException:
 | 
						|
            logger.exception("webdriver quit was not successful")
 | 
						|
        except percy.errors.Error:  # type: ignore[reportAttributeAccessIssue]
 | 
						|
            logger.exception("percy runner failed to finalize properly")
 | 
						|
 | 
						|
    def visit_and_snapshot(
 | 
						|
        self,
 | 
						|
        resource_path: str,
 | 
						|
        hook_id: str,
 | 
						|
        wait_for_callbacks=True,
 | 
						|
        convert_canvases=False,
 | 
						|
        assert_check=True,
 | 
						|
        stay_on_page=False,
 | 
						|
        widths=None,
 | 
						|
    ):
 | 
						|
        path = resource_path.lstrip("/")
 | 
						|
        try:
 | 
						|
            if path != resource_path:
 | 
						|
                logger.warning("we stripped the left '/' in resource_path")
 | 
						|
            self.server_url = self.server_url
 | 
						|
            self.driver.get(f"{self.server_url.rstrip('/')}/{path}")
 | 
						|
 | 
						|
            # wait for the hook_id to present and all callbacks get fired
 | 
						|
            self.wait_for_element_by_id(hook_id)
 | 
						|
            self.percy_snapshot(
 | 
						|
                path,
 | 
						|
                wait_for_callbacks=wait_for_callbacks,
 | 
						|
                convert_canvases=convert_canvases,
 | 
						|
                widths=widths,
 | 
						|
            )
 | 
						|
            if assert_check:
 | 
						|
                assert not self.find_elements(
 | 
						|
                    "div.dash-debug-alert"
 | 
						|
                ), "devtools should not raise an error alert"
 | 
						|
            if not stay_on_page:
 | 
						|
                self.driver.back()
 | 
						|
        except WebDriverException as e:
 | 
						|
            logger.exception("snapshot at resource %s error", path)
 | 
						|
            raise e
 | 
						|
 | 
						|
    def percy_snapshot(
 | 
						|
        self, name="", wait_for_callbacks=False, convert_canvases=False, widths=None
 | 
						|
    ):
 | 
						|
        """percy_snapshot - visual test api shortcut to `percy_runner.snapshot`.
 | 
						|
        It also combines the snapshot `name` with the Python version,
 | 
						|
        args:
 | 
						|
        - name: combined with the python version to give the final snapshot name
 | 
						|
        - wait_for_callbacks: default False, whether to wait for Dash callbacks,
 | 
						|
            after an extra second to ensure that any relevant callbacks have
 | 
						|
            been initiated
 | 
						|
        - convert_canvases: default False, whether to convert all canvas elements
 | 
						|
            in the DOM into static images for percy to see. They will be restored
 | 
						|
            after the snapshot is complete.
 | 
						|
        - widths: a list of pixel widths for percy to render the page with. Note
 | 
						|
            that this does not change the browser in which the DOM is constructed,
 | 
						|
            so the width will only affect CSS, not JS-driven layout.
 | 
						|
            Defaults to [1280]
 | 
						|
        """
 | 
						|
        if widths is None:
 | 
						|
            widths = [1280]
 | 
						|
        try:
 | 
						|
            import asgiref  # pylint: disable=unused-import, import-outside-toplevel # noqa: F401, C0415
 | 
						|
 | 
						|
            name += "_async"
 | 
						|
        except ImportError:
 | 
						|
            pass
 | 
						|
 | 
						|
        logger.info("taking snapshot name => %s", name)
 | 
						|
        try:
 | 
						|
            if wait_for_callbacks:
 | 
						|
                # the extra one second sleep adds safe margin in the context
 | 
						|
                # of wait_for_callbacks
 | 
						|
                time.sleep(1)
 | 
						|
                until(self._wait_for_callbacks, timeout=40, poll=0.3)
 | 
						|
        except TestingTimeoutError:
 | 
						|
            # API will log the error but this TimeoutError should not block
 | 
						|
            # the test execution to continue and it will still do a snapshot
 | 
						|
            # as diff reference for the build run.
 | 
						|
            logger.error(
 | 
						|
                "wait_for_callbacks failed => status of invalid rqs %s",
 | 
						|
                self.redux_state_rqs,
 | 
						|
            )
 | 
						|
 | 
						|
        if convert_canvases:
 | 
						|
            self.driver.execute_script(
 | 
						|
                """
 | 
						|
                const stash = window._canvasStash = [];
 | 
						|
                Array.from(document.querySelectorAll('canvas')).forEach(c => {
 | 
						|
                    const i = document.createElement('img');
 | 
						|
                    i.src = c.toDataURL();
 | 
						|
                    i.width = c.width;
 | 
						|
                    i.height = c.height;
 | 
						|
                    i.setAttribute('style', c.getAttribute('style'));
 | 
						|
                    i.className = c.className;
 | 
						|
                    i.setAttribute('data-canvasnum', stash.length);
 | 
						|
                    stash.push(c);
 | 
						|
                    c.parentElement.insertBefore(i, c);
 | 
						|
                    c.parentElement.removeChild(c);
 | 
						|
                });
 | 
						|
            """
 | 
						|
            )
 | 
						|
 | 
						|
        try:
 | 
						|
            self.percy_runner.snapshot(name=name, widths=widths)
 | 
						|
        except requests.HTTPError as err:
 | 
						|
            # Ignore retries.
 | 
						|
            if err.request.status_code != 400:  # type: ignore[reportAttributeAccessIssue]
 | 
						|
                raise err
 | 
						|
 | 
						|
        if convert_canvases:
 | 
						|
            self.driver.execute_script(
 | 
						|
                """
 | 
						|
                const stash = window._canvasStash;
 | 
						|
                Array.from(
 | 
						|
                    document.querySelectorAll('img[data-canvasnum]')
 | 
						|
                ).forEach(i => {
 | 
						|
                    const c = stash[+i.getAttribute('data-canvasnum')];
 | 
						|
                    i.parentElement.insertBefore(c, i);
 | 
						|
                    i.parentElement.removeChild(i);
 | 
						|
                });
 | 
						|
                delete window._canvasStash;
 | 
						|
            """
 | 
						|
            )
 | 
						|
 | 
						|
    def take_snapshot(self, name: str):
 | 
						|
        """Hook method to take snapshot when a selenium test fails. The
 | 
						|
        snapshot is placed under.
 | 
						|
 | 
						|
            - `/tmp/dash_artifacts` in linux
 | 
						|
            - `%TEMP` in windows
 | 
						|
        with a filename combining test case name and the
 | 
						|
        running selenium session id
 | 
						|
        """
 | 
						|
        target = (
 | 
						|
            "/tmp/dash_artifacts" if not self._is_windows() else os.getenv("TEMP", "")
 | 
						|
        )
 | 
						|
 | 
						|
        if not os.path.exists(target):
 | 
						|
            try:
 | 
						|
                os.mkdir(target)
 | 
						|
            except OSError:
 | 
						|
                logger.exception("cannot make artifacts")
 | 
						|
 | 
						|
        self.driver.save_screenshot(f"{target}/{name}_{self.session_id}.png")
 | 
						|
 | 
						|
    def find_element(self, selector, attribute="CSS_SELECTOR"):
 | 
						|
        """find_element returns the first found element by the attribute `selector`
 | 
						|
        shortcut to `driver.find_element(By.CSS_SELECTOR, ...)`.
 | 
						|
        args:
 | 
						|
        - attribute: the attribute type to search for, aligns with the Selenium
 | 
						|
            API's `By` class. default "CSS_SELECTOR"
 | 
						|
            valid values: "CSS_SELECTOR", "ID", "NAME", "TAG_NAME",
 | 
						|
            "CLASS_NAME", "LINK_TEXT", "PARTIAL_LINK_TEXT", "XPATH"
 | 
						|
        """
 | 
						|
        return self.driver.find_element(getattr(By, attribute.upper()), selector)
 | 
						|
 | 
						|
    def find_elements(self, selector, attribute="CSS_SELECTOR"):
 | 
						|
        """find_elements returns a list of all elements matching the attribute
 | 
						|
        `selector`. Shortcut to `driver.find_elements(By.CSS_SELECTOR, ...)`.
 | 
						|
        args:
 | 
						|
        - attribute: the attribute type to search for, aligns with the Selenium
 | 
						|
            API's `By` class. default "CSS_SELECTOR"
 | 
						|
            valid values: "CSS_SELECTOR", "ID", "NAME", "TAG_NAME",
 | 
						|
            "CLASS_NAME", "LINK_TEXT", "PARTIAL_LINK_TEXT", "XPATH"
 | 
						|
        """
 | 
						|
        return self.driver.find_elements(getattr(By, attribute.upper()), selector)
 | 
						|
 | 
						|
    def _get_element(self, elem_or_selector):
 | 
						|
        if isinstance(elem_or_selector, str):
 | 
						|
            return self.find_element(elem_or_selector)
 | 
						|
        return elem_or_selector
 | 
						|
 | 
						|
    def _wait_for(self, method, timeout, msg):
 | 
						|
        """Abstract generic pattern for explicit WebDriverWait."""
 | 
						|
        try:
 | 
						|
            _wait = (
 | 
						|
                self._wd_wait
 | 
						|
                if timeout is None
 | 
						|
                else WebDriverWait(self.driver, timeout)
 | 
						|
            )
 | 
						|
            logger.debug(
 | 
						|
                "method, timeout, poll => %s %s %s",
 | 
						|
                method,
 | 
						|
                _wait._timeout,  # pylint: disable=protected-access
 | 
						|
                _wait._poll,  # pylint: disable=protected-access
 | 
						|
            )
 | 
						|
 | 
						|
            return _wait.until(method)
 | 
						|
        except Exception as err:
 | 
						|
            if callable(msg):
 | 
						|
                message = msg(self.driver)
 | 
						|
            else:
 | 
						|
                message = msg
 | 
						|
            raise TimeoutException(str(message)) from err
 | 
						|
 | 
						|
    def wait_for_element(self, selector, timeout=None):
 | 
						|
        """wait_for_element is shortcut to `wait_for_element_by_css_selector`
 | 
						|
        timeout if not set, equals to the fixture's `wait_timeout`."""
 | 
						|
        return self.wait_for_element_by_css_selector(selector, timeout)
 | 
						|
 | 
						|
    def wait_for_element_by_css_selector(self, selector, timeout=None):
 | 
						|
        """Explicit wait until the element is present, timeout if not set,
 | 
						|
        equals to the fixture's `wait_timeout` shortcut to `WebDriverWait` with
 | 
						|
        `EC.presence_of_element_located`."""
 | 
						|
        return self._wait_for(
 | 
						|
            EC.presence_of_element_located(
 | 
						|
                (By.CSS_SELECTOR, selector),
 | 
						|
            ),
 | 
						|
            timeout,
 | 
						|
            f"timeout {timeout or self._wait_timeout}s => waiting for selector {selector}",
 | 
						|
        )
 | 
						|
 | 
						|
    def wait_for_no_elements(self, selector, timeout=None):
 | 
						|
        """Explicit wait until an element is NOT found. timeout defaults to
 | 
						|
        the fixture's `wait_timeout`."""
 | 
						|
        until(
 | 
						|
            # if we use get_elements it waits a long time to see if they appear
 | 
						|
            # so this one calls out directly to execute_script
 | 
						|
            lambda: self.driver.execute_script(
 | 
						|
                f"return document.querySelectorAll('{selector}').length"
 | 
						|
            )
 | 
						|
            == 0,
 | 
						|
            timeout or self._wait_timeout,
 | 
						|
        )
 | 
						|
 | 
						|
    def wait_for_element_by_id(self, element_id, timeout=None):
 | 
						|
        """Explicit wait until the element is present, timeout if not set,
 | 
						|
        equals to the fixture's `wait_timeout` shortcut to `WebDriverWait` with
 | 
						|
        `EC.presence_of_element_located`."""
 | 
						|
        return self._wait_for(
 | 
						|
            EC.presence_of_element_located(
 | 
						|
                (By.ID, element_id),
 | 
						|
            ),
 | 
						|
            timeout,
 | 
						|
            f"timeout {timeout or self._wait_timeout}s => waiting for element id {element_id}",
 | 
						|
        )
 | 
						|
 | 
						|
    def wait_for_class_to_equal(self, selector, classname, timeout=None):
 | 
						|
        """Explicit wait until the element's class has expected `value` timeout
 | 
						|
        if not set, equals to the fixture's `wait_timeout` shortcut to
 | 
						|
        `WebDriverWait` with customized `class_to_equal` condition."""
 | 
						|
        return self._wait_for(
 | 
						|
            method=class_to_equal(selector, classname),
 | 
						|
            timeout=timeout,
 | 
						|
            msg=f"classname => {classname} not found within {timeout or self._wait_timeout}s",
 | 
						|
        )
 | 
						|
 | 
						|
    def wait_for_style_to_equal(self, selector, style, val, timeout=None):
 | 
						|
        """Explicit wait until the element's style has expected `value` timeout
 | 
						|
        if not set, equals to the fixture's `wait_timeout` shortcut to
 | 
						|
        `WebDriverWait` with customized `style_to_equal` condition."""
 | 
						|
        return self._wait_for(
 | 
						|
            method=style_to_equal(selector, style, val),
 | 
						|
            timeout=timeout,
 | 
						|
            msg=f"style val => {style} {val} not found within {timeout or self._wait_timeout}s",
 | 
						|
        )
 | 
						|
 | 
						|
    def wait_for_text_to_equal(self, selector, text, timeout=None):
 | 
						|
        """Explicit wait until the element's text equals the expected `text`.
 | 
						|
 | 
						|
        timeout if not set, equals to the fixture's `wait_timeout`
 | 
						|
        shortcut to `WebDriverWait` with customized `text_to_equal`
 | 
						|
        condition.
 | 
						|
        """
 | 
						|
        method = text_to_equal(selector, text, timeout or self.wait_timeout)
 | 
						|
 | 
						|
        return self._wait_for(
 | 
						|
            method=method,
 | 
						|
            timeout=timeout,
 | 
						|
            msg=method.message,
 | 
						|
        )
 | 
						|
 | 
						|
    def wait_for_contains_class(self, selector, classname, timeout=None):
 | 
						|
        """Explicit wait until the element's classes contains the expected `classname`.
 | 
						|
 | 
						|
        timeout if not set, equals to the fixture's `wait_timeout`
 | 
						|
        shortcut to `WebDriverWait` with customized `contains_class`
 | 
						|
        condition.
 | 
						|
        """
 | 
						|
        return self._wait_for(
 | 
						|
            method=contains_class(selector, classname),
 | 
						|
            timeout=timeout,
 | 
						|
            msg=f"classname -> {classname} not found inside element within {timeout or self._wait_timeout}s",
 | 
						|
        )
 | 
						|
 | 
						|
    def wait_for_contains_text(self, selector, text, timeout=None):
 | 
						|
        """Explicit wait until the element's text contains the expected `text`.
 | 
						|
 | 
						|
        timeout if not set, equals to the fixture's `wait_timeout`
 | 
						|
        shortcut to `WebDriverWait` with customized `contains_text`
 | 
						|
        condition.
 | 
						|
        """
 | 
						|
        method = contains_text(selector, text, timeout or self.wait_timeout)
 | 
						|
        return self._wait_for(
 | 
						|
            method=method,
 | 
						|
            timeout=timeout,
 | 
						|
            msg=method.message,
 | 
						|
        )
 | 
						|
 | 
						|
    def wait_for_page(self, url=None, timeout=10):
 | 
						|
        """wait_for_page navigates to the url in webdriver wait until the
 | 
						|
        renderer is loaded in browser.
 | 
						|
 | 
						|
        use the `server_url` if url is not provided.
 | 
						|
        """
 | 
						|
        self.driver.get(self.server_url if url is None else url)
 | 
						|
        try:
 | 
						|
            self.wait_for_element_by_css_selector(
 | 
						|
                self.dash_entry_locator, timeout=timeout
 | 
						|
            )
 | 
						|
        except TimeoutException as exc:
 | 
						|
            logger.exception("dash server is not loaded within %s seconds", timeout)
 | 
						|
            logs = "\n".join((str(log) for log in self.get_logs()))  # type: ignore[reportOptionalIterable]
 | 
						|
            logger.debug(logs)
 | 
						|
            html = self.find_element("body").get_property("innerHTML")
 | 
						|
            raise DashAppLoadingError(
 | 
						|
                "the expected Dash react entry point cannot be loaded"
 | 
						|
                f" in browser\n HTML => {html}\n Console Logs => {logs}\n"
 | 
						|
            ) from exc
 | 
						|
 | 
						|
        if self._pause:
 | 
						|
            import pdb  # pylint: disable=import-outside-toplevel
 | 
						|
 | 
						|
            pdb.set_trace()  # pylint: disable=forgotten-debug-statement
 | 
						|
 | 
						|
    def select_dcc_dropdown(self, elem_or_selector, value=None, index=None):
 | 
						|
        dropdown = self._get_element(elem_or_selector)
 | 
						|
        dropdown.click()
 | 
						|
 | 
						|
        menu = dropdown.find_element(By.CSS_SELECTOR, "div.Select-menu-outer")
 | 
						|
        logger.debug("the available options are %s", "|".join(menu.text.split("\n")))
 | 
						|
 | 
						|
        options = menu.find_elements(By.CSS_SELECTOR, "div.VirtualizedSelectOption")
 | 
						|
        if options:
 | 
						|
            if isinstance(index, int):
 | 
						|
                options[index].click()
 | 
						|
                return
 | 
						|
 | 
						|
            for option in options:
 | 
						|
                if option.text == value:
 | 
						|
                    option.click()
 | 
						|
                    return
 | 
						|
 | 
						|
        logger.error(
 | 
						|
            "cannot find matching option using value=%s or index=%s", value, index
 | 
						|
        )
 | 
						|
 | 
						|
    def toggle_window(self):
 | 
						|
        """Switch between the current working window and the new opened one."""
 | 
						|
        idx = (self._window_idx + 1) % 2
 | 
						|
        self.switch_window(idx=idx)
 | 
						|
        self._window_idx += 1
 | 
						|
 | 
						|
    def switch_window(self, idx=0):
 | 
						|
        """Switch to window by window index shortcut to
 | 
						|
        `driver.switch_to.window`."""
 | 
						|
        if len(self.driver.window_handles) <= idx:
 | 
						|
            raise BrowserError("there is no second window in Browser")
 | 
						|
 | 
						|
        self.driver.switch_to.window(self.driver.window_handles[idx])
 | 
						|
 | 
						|
    def open_new_tab(self, url=None):
 | 
						|
        """Open a new tab in browser url is not set, equals to `server_url`."""
 | 
						|
        self.driver.execute_script(
 | 
						|
            f'window.open("{url or self.server_url}", "new window")'
 | 
						|
        )
 | 
						|
 | 
						|
    def get_webdriver(self):
 | 
						|
        return getattr(self, f"_get_{self._browser}")()
 | 
						|
 | 
						|
    def _get_wd_options(self):
 | 
						|
        options = (
 | 
						|
            self._options[0]
 | 
						|
            if self._options and isinstance(self._options, list)
 | 
						|
            else getattr(webdriver, self._browser).options.Options()
 | 
						|
        )
 | 
						|
 | 
						|
        if self._headless:
 | 
						|
            options.add_argument("--headless")
 | 
						|
 | 
						|
        return options
 | 
						|
 | 
						|
    def _get_chrome(self):
 | 
						|
        options = self._get_wd_options()
 | 
						|
 | 
						|
        if "DASH_TEST_CHROMEPATH" in os.environ:
 | 
						|
            options.binary_location = os.environ["DASH_TEST_CHROMEPATH"]
 | 
						|
 | 
						|
        options.add_experimental_option(
 | 
						|
            "prefs",
 | 
						|
            {
 | 
						|
                "download.default_directory": self.download_path,
 | 
						|
                "download.prompt_for_download": False,
 | 
						|
                "download.directory_upgrade": True,
 | 
						|
                "safebrowsing.enabled": False,
 | 
						|
                "safebrowsing.disable_download_protection": True,
 | 
						|
            },
 | 
						|
        )
 | 
						|
 | 
						|
        options.add_argument("--disable-dev-shm-usage")
 | 
						|
        options.add_argument("--no-sandbox")
 | 
						|
        options.add_argument("--disable-gpu")
 | 
						|
        options.add_argument("--remote-debugging-port=0")
 | 
						|
 | 
						|
        options.set_capability("goog:loggingPrefs", {"browser": "SEVERE"})
 | 
						|
 | 
						|
        chrome = (
 | 
						|
            webdriver.Remote(command_executor=self._remote_url, options=options)  # type: ignore[reportAttributeAccessIssue]
 | 
						|
            if self._remote
 | 
						|
            else webdriver.Chrome(options=options)
 | 
						|
        )
 | 
						|
 | 
						|
        # https://bugs.chromium.org/p/chromium/issues/detail?id=696481
 | 
						|
        if self._headless:
 | 
						|
            # pylint: disable=protected-access
 | 
						|
            chrome.command_executor._commands["send_command"] = (  # type: ignore[reportArgumentType]
 | 
						|
                "POST",
 | 
						|
                "/session/$sessionId/chromium/send_command",
 | 
						|
            )
 | 
						|
            params = {
 | 
						|
                "cmd": "Page.setDownloadBehavior",
 | 
						|
                "params": {"behavior": "allow", "downloadPath": self.download_path},
 | 
						|
            }
 | 
						|
            res = chrome.execute("send_command", params)
 | 
						|
            logger.debug("enabled headless download returns %s", res)
 | 
						|
 | 
						|
        chrome.set_window_position(0, 0)
 | 
						|
        return chrome
 | 
						|
 | 
						|
    def _get_firefox(self):
 | 
						|
        options = self._get_wd_options()
 | 
						|
 | 
						|
        options.set_capability("marionette", True)
 | 
						|
 | 
						|
        options.set_preference("browser.download.dir", self.download_path)
 | 
						|
        options.set_preference("browser.download.folderList", 2)
 | 
						|
        options.set_preference(
 | 
						|
            "browser.helperApps.neverAsk.saveToDisk",
 | 
						|
            "application/octet-stream",  # this MIME is generic for binary
 | 
						|
        )
 | 
						|
        if not self._remote_url and self._remote:
 | 
						|
            raise TypeError("remote_url was not provided but required for Firefox")
 | 
						|
 | 
						|
        return (
 | 
						|
            webdriver.Remote(
 | 
						|
                command_executor=self._remote_url,  # type: ignore[reportTypeArgument]
 | 
						|
                options=options,
 | 
						|
            )
 | 
						|
            if self._remote
 | 
						|
            else webdriver.Firefox(options=options)
 | 
						|
        )
 | 
						|
 | 
						|
    @staticmethod
 | 
						|
    def _is_windows():
 | 
						|
        return sys.platform == "win32"
 | 
						|
 | 
						|
    def multiple_click(self, elem_or_selector, clicks, delay=None):
 | 
						|
        """multiple_click click the element with number of `clicks`."""
 | 
						|
        for _ in range(clicks):
 | 
						|
            self._get_element(elem_or_selector).click()
 | 
						|
            if delay:
 | 
						|
                time.sleep(delay)
 | 
						|
 | 
						|
    def clear_input(self, elem_or_selector):
 | 
						|
        """Simulate key press to clear the input."""
 | 
						|
        elem = self._get_element(elem_or_selector)
 | 
						|
        logger.debug("clear input with %s => %s", elem_or_selector, elem)
 | 
						|
        (
 | 
						|
            ActionChains(self.driver)
 | 
						|
            .move_to_element(elem)
 | 
						|
            .pause(0.2)
 | 
						|
            .click(elem)
 | 
						|
            .send_keys(Keys.END)
 | 
						|
            .key_down(Keys.SHIFT)
 | 
						|
            .send_keys(Keys.HOME)
 | 
						|
            .key_up(Keys.SHIFT)
 | 
						|
            .send_keys(Keys.DELETE)
 | 
						|
        ).perform()
 | 
						|
 | 
						|
    def zoom_in_graph_by_ratio(
 | 
						|
        self, elem_or_selector, start_fraction=0.5, zoom_box_fraction=0.2, compare=True
 | 
						|
    ):
 | 
						|
        """Zoom out a graph with a zoom box fraction of component dimension
 | 
						|
        default start at middle with a rectangle of 1/5 of the dimension use
 | 
						|
        `compare` to control if we check the svg get changed."""
 | 
						|
        elem = self._get_element(elem_or_selector)
 | 
						|
 | 
						|
        prev = elem.get_attribute("innerHTML")
 | 
						|
        w, h = elem.size["width"], elem.size["height"]
 | 
						|
        try:
 | 
						|
            ActionChains(self.driver).move_to_element_with_offset(
 | 
						|
                elem, w * start_fraction, h * start_fraction
 | 
						|
            ).drag_and_drop_by_offset(
 | 
						|
                elem, w * zoom_box_fraction, h * zoom_box_fraction
 | 
						|
            ).perform()
 | 
						|
        except MoveTargetOutOfBoundsException:
 | 
						|
            logger.exception("graph offset outside of the boundary")
 | 
						|
        if compare:
 | 
						|
            assert prev != elem.get_attribute(
 | 
						|
                "innerHTML"
 | 
						|
            ), "SVG content should be different after zoom"
 | 
						|
 | 
						|
    def click_at_coord_fractions(self, elem_or_selector, fx, fy):
 | 
						|
        elem = self._get_element(elem_or_selector)
 | 
						|
 | 
						|
        ActionChains(self.driver).move_to_element_with_offset(
 | 
						|
            elem, elem.size["width"] * fx, elem.size["height"] * fy
 | 
						|
        ).click().perform()
 | 
						|
 | 
						|
    def get_logs(self):
 | 
						|
        """Return a list of `SEVERE` level logs after last reset time stamps
 | 
						|
        (default to 0, resettable by `reset_log_timestamp`.
 | 
						|
 | 
						|
        Chrome only
 | 
						|
        """
 | 
						|
        if self._browser == "chrome":
 | 
						|
            return [
 | 
						|
                entry
 | 
						|
                for entry in self.driver.get_log("browser")
 | 
						|
                if entry["timestamp"] > self._last_ts
 | 
						|
            ]
 | 
						|
        warnings.warn("get_logs always return None with webdrivers other than Chrome")
 | 
						|
        return None
 | 
						|
 | 
						|
    def reset_log_timestamp(self):
 | 
						|
        """reset_log_timestamp only work with chrome webdriver."""
 | 
						|
        if self._browser == "chrome":
 | 
						|
            entries = self.driver.get_log("browser")
 | 
						|
            if entries:
 | 
						|
                self._last_ts = entries[-1]["timestamp"]
 | 
						|
 | 
						|
    @property
 | 
						|
    def driver(self):
 | 
						|
        """Expose the selenium webdriver as fixture property."""
 | 
						|
        return self._driver
 | 
						|
 | 
						|
    @property
 | 
						|
    def session_id(self):
 | 
						|
        return self.driver.session_id
 | 
						|
 | 
						|
    @property
 | 
						|
    def server_url(self) -> str:
 | 
						|
        return self._url
 | 
						|
 | 
						|
    @server_url.setter
 | 
						|
    def server_url(self, value):
 | 
						|
        """Set the server url so the selenium is aware of the local server
 | 
						|
        port.
 | 
						|
 | 
						|
        It also implicitly calls `wait_for_page`.
 | 
						|
        """
 | 
						|
        self._url = value
 | 
						|
        self.wait_for_page()
 | 
						|
 | 
						|
    @property
 | 
						|
    def download_path(self):
 | 
						|
        return self._download_path
 | 
						|
 | 
						|
    @property
 | 
						|
    def wait_timeout(self):
 | 
						|
        return self._wait_timeout
 | 
						|
 | 
						|
    @wait_timeout.setter
 | 
						|
    def wait_timeout(self, value):
 | 
						|
        self._wait_timeout = value
 | 
						|
        self._wd_wait = WebDriverWait(self.driver, value)
 |