Warning
This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.
Source code for galaxy_test.selenium.framework
"""Basis for Selenium test framework."""
import datetime
import json
import os
import traceback
import unittest
from functools import partial, wraps
import requests
from gxformat2 import (
convert_and_import_workflow,
ImporterGalaxyInterface,
)
try:
from pyvirtualdisplay import Display
except ImportError:
Display = None
from galaxy.selenium import (
driver_factory,
)
from galaxy.selenium.context import (
GalaxySeleniumContext,
)
from galaxy.selenium.navigates_galaxy import (
retry_during_transitions
)
from galaxy.util import asbool, classproperty
from galaxy_test.base import populators
from galaxy_test.base.api import UsesApiTestCaseMixin
from galaxy_test.base.env import DEFAULT_WEB_HOST, get_ip_address
from galaxy_test.base.testcase import FunctionalTestCase
try:
from galaxy_test.driver.driver_util import GalaxyTestDriver
except ImportError:
GalaxyTestDriver = None # type: ignore
DEFAULT_TIMEOUT_MULTIPLIER = 1
DEFAULT_TEST_ERRORS_DIRECTORY = os.path.abspath("database/test_errors")
DEFAULT_SELENIUM_HEADLESS = "auto"
DEFAULT_ADMIN_USER = "test@bx.psu.edu"
DEFAULT_ADMIN_PASSWORD = "testpass"
DEFAULT_DOWNLOAD_PATH = driver_factory.DEFAULT_DOWNLOAD_PATH
TIMEOUT_MULTIPLIER = float(os.environ.get("GALAXY_TEST_TIMEOUT_MULTIPLIER", DEFAULT_TIMEOUT_MULTIPLIER))
GALAXY_TEST_ERRORS_DIRECTORY = os.environ.get("GALAXY_TEST_ERRORS_DIRECTORY", DEFAULT_TEST_ERRORS_DIRECTORY)
GALAXY_TEST_SCREENSHOTS_DIRECTORY = os.environ.get("GALAXY_TEST_SCREENSHOTS_DIRECTORY", None)
# Test browser can be ["CHROME", "FIREFOX", "OPERA", "PHANTOMJS"]
GALAXY_TEST_SELENIUM_BROWSER = os.environ.get("GALAXY_TEST_SELENIUM_BROWSER", driver_factory.DEFAULT_SELENIUM_BROWSER)
GALAXY_TEST_SELENIUM_REMOTE = os.environ.get("GALAXY_TEST_SELENIUM_REMOTE", driver_factory.DEFAULT_SELENIUM_REMOTE)
GALAXY_TEST_SELENIUM_REMOTE_PORT = os.environ.get("GALAXY_TEST_SELENIUM_REMOTE_PORT", driver_factory.DEFAULT_SELENIUM_REMOTE_PORT)
GALAXY_TEST_SELENIUM_REMOTE_HOST = os.environ.get("GALAXY_TEST_SELENIUM_REMOTE_HOST", driver_factory.DEFAULT_SELENIUM_REMOTE_HOST)
GALAXY_TEST_SELENIUM_HEADLESS = os.environ.get("GALAXY_TEST_SELENIUM_HEADLESS", DEFAULT_SELENIUM_HEADLESS)
GALAXY_TEST_EXTERNAL_FROM_SELENIUM = os.environ.get("GALAXY_TEST_EXTERNAL_FROM_SELENIUM", None)
# Auto-retry selenium tests this many times.
GALAXY_TEST_SELENIUM_RETRIES = int(os.environ.get("GALAXY_TEST_SELENIUM_RETRIES", "0"))
GALAXY_TEST_SELENIUM_USER_EMAIL = os.environ.get("GALAXY_TEST_SELENIUM_USER_EMAIL", None)
GALAXY_TEST_SELENIUM_USER_PASSWORD = os.environ.get("GALAXY_TEST_SELENIUM_USER_PASSWORD", None)
GALAXY_TEST_SELENIUM_ADMIN_USER_EMAIL = os.environ.get("GALAXY_TEST_SELENIUM_ADMIN_USER_EMAIL", DEFAULT_ADMIN_USER)
GALAXY_TEST_SELENIUM_ADMIN_USER_PASSWORD = os.environ.get("GALAXY_TEST_SELENIUM_ADMIN_USER_PASSWORD", DEFAULT_ADMIN_PASSWORD)
# JS code to execute in Galaxy JS console to setup localStorage of session for logging and
# logging "flatten" messages because it seems Selenium (with Chrome at least) only grabs
# the first argument to console.XXX when recovering the browser log.
SETUP_LOGGING_JS = '''
window.localStorage && window.localStorage.setItem("galaxy:debug", true);
window.localStorage && window.localStorage.setItem("galaxy:debug:flatten", true);
'''
try:
from nose.tools import nottest
except ImportError:
def nottest(x):
return x
[docs]def managed_history(f):
"""Ensure a Selenium test has a distinct, named history.
Cleanup the history after the job is complete as well unless
GALAXY_TEST_NO_CLEANUP is set in the environment.
"""
@wraps(f)
def func_wrapper(self, *args, **kwds):
self.home()
history_name = f.__name__ + datetime.datetime.now().strftime("%Y%m%d%H%M%s")
self.history_panel_create_new_with_name(history_name)
try:
f(self, *args, **kwds)
finally:
if "GALAXY_TEST_NO_CLEANUP" not in os.environ:
try:
current_history_id = self.current_history_id()
self.dataset_populator.cancel_history_jobs(current_history_id)
self.api_delete("histories/%s" % current_history_id)
except Exception:
print("Faild to cleanup managed history, selenium connection corrupted somehow?")
return func_wrapper
[docs]def dump_test_information(self, name_prefix):
if GALAXY_TEST_ERRORS_DIRECTORY and GALAXY_TEST_ERRORS_DIRECTORY != "0":
if not os.path.exists(GALAXY_TEST_ERRORS_DIRECTORY):
os.makedirs(GALAXY_TEST_ERRORS_DIRECTORY)
result_name = name_prefix + datetime.datetime.now().strftime("%Y%m%d%H%M%s")
target_directory = os.path.join(GALAXY_TEST_ERRORS_DIRECTORY, result_name)
def write_file(name, content, raw=False):
with open(os.path.join(target_directory, name), "wb") as buf:
buf.write(content.encode("utf-8") if not raw else content)
os.makedirs(target_directory)
write_file("stacktrace.txt", traceback.format_exc())
for snapshot in getattr(self, "snapshots", []):
snapshot.write_to_error_directory(write_file)
# Try to use the Selenium driver to recover more debug information, but don't
# throw an exception if the connection is broken in some way.
try:
self.driver.save_screenshot(os.path.join(target_directory, "last.png"))
write_file("page_source.txt", self.driver.page_source)
write_file("DOM.txt", self.driver.execute_script("return document.documentElement.outerHTML"))
except Exception:
print("Failed to use test driver to recover debug information from Selenium.")
write_file("selenium_exception.txt", traceback.format_exc())
for log_type in ["browser", "driver"]:
try:
full_log = self.driver.get_log(log_type)
trimmed_log = [entry for entry in full_log if entry["level"] not in ["DEBUG", "INFO"]]
write_file("%s.log.json" % log_type, json.dumps(trimmed_log, indent=True))
write_file("%s.log.verbose.json" % log_type, json.dumps(full_log, indent=True))
except Exception:
continue
[docs]@nottest
def selenium_test(f):
test_name = f.__name__
@wraps(f)
def func_wrapper(self, *args, **kwds):
retry_attempts = 0
while True:
if retry_attempts > 0:
self.reset_driver_and_session()
try:
return f(self, *args, **kwds)
except unittest.SkipTest:
dump_test_information(self, test_name)
# Don't retry if we have purposely decided to skip the test.
raise
except Exception:
dump_test_information(self, test_name)
if retry_attempts < GALAXY_TEST_SELENIUM_RETRIES:
retry_attempts += 1
print(f"Test function [{test_name}] threw an exception, retrying. Failed attempts - {retry_attempts}.")
else:
raise
return func_wrapper
retry_assertion_during_transitions = partial(retry_during_transitions, exception_check=lambda e: isinstance(e, AssertionError))
[docs]class TestSnapshot:
[docs] def __init__(self, driver, index, description):
self.screenshot_binary = driver.get_screenshot_as_png()
self.description = description
self.index = index
self.exc = traceback.format_exc()
self.stack = traceback.format_stack()
[docs] def write_to_error_directory(self, write_file_func):
prefix = "%d-%s" % (self.index, self.description)
write_file_func("%s-screenshot.png" % prefix, self.screenshot_binary, raw=True)
write_file_func("%s-traceback.txt" % prefix, self.exc)
write_file_func("%s-stack.txt" % prefix, str(self.stack))
[docs]class TestWithSeleniumMixin(GalaxySeleniumContext, UsesApiTestCaseMixin):
# If run one-off via nosetests, the next line ensures test
# tools and datatypes are used instead of configured tools.
framework_tool_and_types = True
# Override this in subclasses to ensure a user is logged in
# before each test. If GALAXY_TEST_SELENIUM_USER_EMAIL and
# GALAXY_TEST_SELENIUM_USER_PASSWORD are set these values
# will be used to login.
ensure_registered = False
# Override this in subclasses to annotate that an admin user
# is required for the test to run properly. Override admin user
# login info with GALAXY_TEST_SELENIUM_ADMIN_USER_EMAIL /
# GALAXY_TEST_SELENIUM_ADMIN_USER_PASSWORD
requires_admin = False
def _target_url_from_selenium(self):
# Deal with the case when Galaxy has a different URL when being accessed by Selenium
# then when being accessed by local API calls.
if GALAXY_TEST_EXTERNAL_FROM_SELENIUM is not None:
target_url_from_selenium = GALAXY_TEST_EXTERNAL_FROM_SELENIUM
else:
target_url_from_selenium = self.url
return target_url_from_selenium
[docs] def setup_selenium(self):
self.target_url_from_selenium = self._target_url_from_selenium()
self.snapshots = []
self.setup_driver_and_session()
if self.requires_admin and GALAXY_TEST_SELENIUM_ADMIN_USER_EMAIL == DEFAULT_ADMIN_USER:
self._setup_interactor()
self._setup_user(GALAXY_TEST_SELENIUM_ADMIN_USER_EMAIL)
self._try_setup_with_driver()
def _try_setup_with_driver(self):
try:
self.setup_with_driver()
except Exception:
dump_test_information(self, self.__class__.__name__ + "_setup")
raise
[docs] def setup_with_driver(self):
"""Override point that allows setting up data using self.driver and Selenium connection.
Overriding this instead of setUp will ensure debug data such as screenshots and stack traces
are dumped if there are problems with the setup and it will be re-ran on test retries.
"""
if self.ensure_registered:
self.login()
[docs] def snapshot(self, description):
"""Create a debug snapshot (DOM, screenshot, etc...) that is written out on tool failure.
This information will be automatically written to a per-test directory created for all
failed tests.
"""
self.snapshots.append(TestSnapshot(self.driver, len(self.snapshots), description))
[docs] def get_download_path(self):
"""Returns default download path
"""
return DEFAULT_DOWNLOAD_PATH
[docs] def api_interactor_for_logged_in_user(self):
api_key = self.get_api_key(force=True)
interactor = self._get_interactor(api_key=api_key)
return interactor
[docs] def write_screenshot_directory_file(self, label, content):
target = self._screenshot_path(label, ".txt")
if target is None:
return
with open(target, "w") as f:
f.write(content)
def _screenshot_path(self, label, extension=".png"):
if GALAXY_TEST_SCREENSHOTS_DIRECTORY is None:
return
if not os.path.exists(GALAXY_TEST_SCREENSHOTS_DIRECTORY):
os.makedirs(GALAXY_TEST_SCREENSHOTS_DIRECTORY)
target = os.path.join(GALAXY_TEST_SCREENSHOTS_DIRECTORY, label + extension)
copy = 1
while os.path.exists(target):
# Maybe previously a test re-run - keep the original.
target = os.path.join(GALAXY_TEST_SCREENSHOTS_DIRECTORY, "%s-%d%s" % (label, copy, extension))
copy += 1
return target
[docs] def reset_driver_and_session(self):
self.tear_down_driver()
self.setup_driver_and_session()
self._try_setup_with_driver()
[docs] def setup_driver_and_session(self):
self.display = driver_factory.virtual_display_if_enabled(use_virtual_display())
self.configured_driver = get_configured_driver()
self._setup_galaxy_logging()
def _setup_galaxy_logging(self):
self.home()
self.driver.execute_script(SETUP_LOGGING_JS)
[docs] def login(self):
if GALAXY_TEST_SELENIUM_USER_EMAIL:
assert GALAXY_TEST_SELENIUM_USER_PASSWORD, "If GALAXY_TEST_SELENIUM_USER_EMAIL is set, a password must be set also with GALAXY_TEST_SELENIUM_USER_PASSWORD"
self.home()
self.submit_login(
email=GALAXY_TEST_SELENIUM_USER_EMAIL,
password=GALAXY_TEST_SELENIUM_USER_PASSWORD,
assert_valid=True,
)
else:
self.register()
[docs] def tear_down_driver(self):
exception = None
try:
self.driver.close()
except Exception as e:
if "cannot kill Chrome" in str(e):
print("Ignoring likely harmless error in Selenium shutdown %s" % e)
else:
exception = e
try:
self.display.stop()
except Exception as e:
exception = e
if exception is not None:
raise exception
@classproperty
def default_web_host(cls):
return default_web_host_for_selenium_tests()
@property
def timeout_multiplier(self):
return TIMEOUT_MULTIPLIER
[docs] def assert_initial_history_panel_state_correct(self):
# Move into a TestsHistoryPanel mixin
unnamed_name = self.components.history_panel.new_name.text
name_element = self.history_panel_name_element()
assert name_element.is_displayed()
assert unnamed_name in name_element.text
initial_size_str = self.components.history_panel.new_size.text
size_selector = self.components.history_panel.size
size_text = size_selector.wait_for_text()
assert initial_size_str in size_text, f"{initial_size_str} not in {size_text}"
self.components.history_panel.empty_message.wait_for_visible()
[docs] def admin_login(self):
self.home()
self.submit_login(
GALAXY_TEST_SELENIUM_ADMIN_USER_EMAIL,
GALAXY_TEST_SELENIUM_ADMIN_USER_PASSWORD
)
with self.main_panel():
self.assert_no_error_message()
@property
def dataset_populator(self):
return SeleniumSessionDatasetPopulator(self)
@property
def dataset_collection_populator(self):
return SeleniumSessionDatasetCollectionPopulator(self)
@property
def workflow_populator(self):
return SeleniumSessionWorkflowPopulator(self)
[docs] def workflow_upload_yaml_with_random_name(self, content, **kwds):
workflow_populator = self.workflow_populator
name = self._get_random_name()
workflow_populator.upload_yaml_workflow(content, name=name, **kwds)
return name
[docs] def ensure_visualization_available(self, hid, visualization_name):
"""Skip or fail a test if visualization for file doesn't appear.
Precondition: viz menu has been opened with history_panel_item_click_visualization_menu.
"""
visualization_names = self.history_panel_item_available_visualizations(hid)
if visualization_name not in visualization_names:
raise unittest.SkipTest("Skipping test, visualization [%s] doesn't appear to be configured." % visualization_name)
[docs]class SeleniumTestCase(FunctionalTestCase, TestWithSeleniumMixin):
galaxy_driver_class = GalaxyTestDriver
[docs] def tearDown(self):
exception = None
try:
self.tear_down_selenium()
except Exception as e:
exception = e
try:
super().tearDown()
except Exception as e:
exception = e
if exception is not None:
raise exception
[docs]class UsesHistoryItemAssertions:
[docs] def assert_item_peek_includes(self, hid, expected):
item_body = self.history_panel_item_component(hid=hid)
peek_text = item_body.peek.wait_for_text()
assert expected in peek_text
[docs] def assert_item_info_includes(self, hid, expected):
item_body = self.history_panel_item_component(hid=hid)
info_text = item_body.info.wait_for_text()
assert expected in info_text, f"Failed to find expected info text [{expected}] in info [{info_text}]"
[docs] def assert_item_dbkey_displayed_as(self, hid, dbkey):
item_body = self.history_panel_item_component(hid=hid)
dbkey_text = item_body.dbkey.wait_for_text()
assert dbkey in dbkey_text
[docs] def assert_item_summary_includes(self, hid, expected_text):
item_body = self.history_panel_item_component(hid=hid)
summary_text = item_body.summary.wait_for_text()
assert expected_text in summary_text, f"Expected summary [{expected_text}] not found in [{summary_text}]."
[docs] def assert_item_name(self, hid, expected_name):
item_body = self.history_panel_item_component(hid=hid)
name = item_body.name.wait_for_text()
assert name == expected_name, name
[docs] def assert_item_hid_text(self, hid):
# Check the text HID matches HID returned from API.
item_body = self.history_panel_item_component(hid=hid)
hid_text = item_body.hid.wait_for_text()
assert hid_text == str(hid), hid_text
[docs]def default_web_host_for_selenium_tests():
if asbool(GALAXY_TEST_SELENIUM_REMOTE):
try:
dev_ip = get_ip_address('docker0')
return dev_ip
except OSError:
return DEFAULT_WEB_HOST
else:
return DEFAULT_WEB_HOST
[docs]def get_configured_driver():
return driver_factory.ConfiguredDriver(
browser=GALAXY_TEST_SELENIUM_BROWSER,
remote=asbool(GALAXY_TEST_SELENIUM_REMOTE),
remote_host=GALAXY_TEST_SELENIUM_REMOTE_HOST,
remote_port=GALAXY_TEST_SELENIUM_REMOTE_PORT,
headless=headless_selenium(),
)
[docs]def headless_selenium():
if asbool(GALAXY_TEST_SELENIUM_REMOTE):
return False
if GALAXY_TEST_SELENIUM_HEADLESS == "auto":
if driver_factory.is_virtual_display_available() or driver_factory.get_local_browser(GALAXY_TEST_SELENIUM_BROWSER) == "CHROME":
return True
else:
return False
else:
return asbool(GALAXY_TEST_SELENIUM_HEADLESS)
[docs]def use_virtual_display():
if asbool(GALAXY_TEST_SELENIUM_REMOTE):
return False
if GALAXY_TEST_SELENIUM_HEADLESS == "auto":
if driver_factory.is_virtual_display_available() and not driver_factory.get_local_browser(GALAXY_TEST_SELENIUM_BROWSER) == "CHROME":
return True
else:
return False
else:
return asbool(GALAXY_TEST_SELENIUM_HEADLESS)
[docs]class SeleniumSessionGetPostMixin:
"""Mixin for adapting Galaxy testing populators helpers to Selenium session backed bioblend."""
def _get(self, route, data=None):
data = data or {}
full_url = self.selenium_test_case.build_url("api/" + route, for_selenium=False)
response = requests.get(full_url, data=data, cookies=self.selenium_test_case.selenium_to_requests_cookies())
return response
def _post(self, route, data=None, files=None):
full_url = self.selenium_test_case.build_url("api/" + route, for_selenium=False)
if data is None:
data = {}
if files is None:
files = data.get("__files", None)
if files is not None:
del data["__files"]
response = requests.post(full_url, data=data, cookies=self.selenium_test_case.selenium_to_requests_cookies(), files=files)
return response
def _delete(self, route, data=None):
data = data or {}
full_url = self.selenium_test_case.build_url("api/" + route, for_selenium=False)
response = requests.delete(full_url, data=data, cookies=self.selenium_test_case.selenium_to_requests_cookies())
return response
def __url(self, route):
return self._gi.url + "/" + route
[docs]class SeleniumSessionDatasetPopulator(populators.BaseDatasetPopulator, SeleniumSessionGetPostMixin):
"""Implementation of BaseDatasetPopulator backed by bioblend."""
[docs] def __init__(self, selenium_test_case):
"""Construct a dataset populator from a bioblend GalaxyInstance."""
self.selenium_test_case = selenium_test_case
[docs]class SeleniumSessionDatasetCollectionPopulator(populators.BaseDatasetCollectionPopulator, SeleniumSessionGetPostMixin):
"""Implementation of BaseDatasetCollectionPopulator backed by bioblend."""
[docs] def __init__(self, selenium_test_case):
"""Construct a dataset collection populator from a bioblend GalaxyInstance."""
self.selenium_test_case = selenium_test_case
self.dataset_populator = SeleniumSessionDatasetPopulator(selenium_test_case)
def _create_collection(self, payload):
create_response = self._post("dataset_collections", data=payload)
return create_response
[docs]class SeleniumSessionWorkflowPopulator(populators.BaseWorkflowPopulator, SeleniumSessionGetPostMixin, ImporterGalaxyInterface):
"""Implementation of BaseWorkflowPopulator backed by bioblend."""
[docs] def __init__(self, selenium_test_case):
"""Construct a workflow populator from a bioblend GalaxyInstance."""
self.selenium_test_case = selenium_test_case
self.dataset_populator = SeleniumSessionDatasetPopulator(selenium_test_case)
[docs] def import_workflow(self, workflow, **kwds):
workflow_str = json.dumps(workflow, indent=4)
data = {
'workflow': workflow_str,
}
data.update(**kwds)
upload_response = self._post("workflows", data=data)
assert upload_response.status_code == 200
return upload_response.json()
[docs] def upload_yaml_workflow(self, has_yaml, **kwds):
workflow = convert_and_import_workflow(has_yaml, galaxy_interface=self, **kwds)
return workflow["id"]