mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-22 06:16:50 +08:00
John Chilton
87bb576c39
These tests aren't idealized unittest.TestCase because they initialize class level data in instance level methods. While this isn't ideal, it is seems an entirely fair workaround given that SeleniumTestCase setups up the Selenium connection itself in an instance method - so class-level initializers would not be able to setup Galaxy data. Since I think we will continue using this pattern then, probably best to formalize it a bit and improve error handling. This provides a formal super class for these test cases that provides a uniform method for setting up the class level data and tracks whether this is successful or not. This serves a couple purposes beyond simple uniformity. First, it tracks if the state has actually been setup or not and will skip subsequent tests if it hasn't. Some of these tests aren't passing very consistently on Jenkins and so we get a bunch of extra noise for tests that are attempting to run without their preconditions met - this will fix that and make the original errors much more clear. Moving the "hacky" part of this into the framework itself also means the tests themselves don't have to repeat hacks like seeing if variables are set with ``getattr`` and such - I always prefer one framework hack to a dozen application hacks.
408 lines
15 KiB
Python
408 lines
15 KiB
Python
"""Basis for Selenium test framework."""
|
|
from __future__ import absolute_import
|
|
from __future__ import print_function
|
|
|
|
import datetime
|
|
import json
|
|
import os
|
|
import time
|
|
import traceback
|
|
import unittest
|
|
|
|
from functools import partial, wraps
|
|
|
|
import requests
|
|
|
|
from galaxy_selenium import (
|
|
driver_factory,
|
|
)
|
|
from galaxy_selenium.navigates_galaxy import NavigatesGalaxy, retry_during_transitions
|
|
|
|
try:
|
|
from pyvirtualdisplay import Display
|
|
except ImportError:
|
|
Display = None
|
|
|
|
from six.moves.urllib.parse import urljoin
|
|
|
|
from base import populators
|
|
from base.driver_util import classproperty, DEFAULT_WEB_HOST, get_ip_address
|
|
from base.testcase import FunctionalTestCase
|
|
from base.workflows_format_2 import (
|
|
ImporterGalaxyInterface,
|
|
convert_and_import_workflow,
|
|
)
|
|
|
|
from galaxy.util import asbool
|
|
|
|
DEFAULT_WAIT_TIMEOUT = 60
|
|
DEFAULT_TEST_ERRORS_DIRECTORY = os.path.abspath("database/test_errors")
|
|
DEFAULT_SELENIUM_BROWSER = "auto"
|
|
DEFAULT_SELENIUM_REMOTE = False
|
|
DEFAULT_SELENIUM_REMOTE_PORT = "4444"
|
|
DEFAULT_SELENIUM_REMOTE_HOST = "127.0.0.1"
|
|
DEFAULT_SELENIUM_HEADLESS = "auto"
|
|
|
|
GALAXY_TEST_ERRORS_DIRECTORY = os.environ.get("GALAXY_TEST_ERRORS_DIRECTORY", DEFAULT_TEST_ERRORS_DIRECTORY)
|
|
# Test browser can be ["CHROME", "FIREFOX", "OPERA", "PHANTOMJS"]
|
|
GALAXY_TEST_SELENIUM_BROWSER = os.environ.get("GALAXY_TEST_SELENIUM_BROWSER", DEFAULT_SELENIUM_BROWSER)
|
|
GALAXY_TEST_SELENIUM_REMOTE = os.environ.get("GALAXY_TEST_SELENIUM_REMOTE", DEFAULT_SELENIUM_REMOTE)
|
|
GALAXY_TEST_SELENIUM_REMOTE_PORT = os.environ.get("GALAXY_TEST_SELENIUM_REMOTE_PORT", DEFAULT_SELENIUM_REMOTE_PORT)
|
|
GALAXY_TEST_SELENIUM_REMOTE_HOST = os.environ.get("GALAXY_TEST_SELENIUM_REMOTE_HOST", 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"))
|
|
|
|
# Test case data
|
|
DEFAULT_PASSWORD = '123456'
|
|
|
|
|
|
try:
|
|
from nose.tools import nottest
|
|
except ImportError:
|
|
def nottest(x):
|
|
return x
|
|
|
|
|
|
@nottest
|
|
def selenium_test(f):
|
|
|
|
@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 Exception:
|
|
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 = f.__name__ + 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):
|
|
with open(os.path.join(target_directory, name), "wb") as buf:
|
|
buf.write(content.encode("utf-8"))
|
|
|
|
os.makedirs(target_directory)
|
|
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"))
|
|
write_file("stacktrace.txt", traceback.format_exc())
|
|
for log_type in ["browser", "driver"]:
|
|
try:
|
|
write_file("%s.log.json" % log_type, json.dumps(self.driver.get_log(log_type)))
|
|
except Exception:
|
|
continue
|
|
iframes = self.driver.find_elements_by_css_selector("iframe")
|
|
for iframe in iframes:
|
|
pass
|
|
# TODO: Dump content out for debugging in the future.
|
|
# iframe_id = iframe.get_attribute("id")
|
|
# if iframe_id:
|
|
# write_file("iframe_%s" % iframe_id, "My content")
|
|
|
|
if retry_attempts < GALAXY_TEST_SELENIUM_RETRIES:
|
|
retry_attempts += 1
|
|
else:
|
|
raise
|
|
|
|
return func_wrapper
|
|
|
|
|
|
retry_assertion_during_transitions = partial(retry_during_transitions, exception_check=lambda e: isinstance(e, AssertionError))
|
|
|
|
|
|
class SeleniumTestCase(FunctionalTestCase, NavigatesGalaxy):
|
|
|
|
framework_tool_and_types = True
|
|
ensure_registered = False
|
|
|
|
def setUp(self):
|
|
super(SeleniumTestCase, self).setUp()
|
|
# 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:
|
|
self.target_url_from_selenium = GALAXY_TEST_EXTERNAL_FROM_SELENIUM
|
|
else:
|
|
self.target_url_from_selenium = self.url
|
|
self.setup_driver_and_session()
|
|
|
|
def tearDown(self):
|
|
exception = None
|
|
try:
|
|
super(SeleniumTestCase, self).tearDown()
|
|
except Exception as e:
|
|
exception = e
|
|
|
|
try:
|
|
self.tear_down_driver()
|
|
except Exception as e:
|
|
exception = e
|
|
|
|
if exception is not None:
|
|
raise exception
|
|
|
|
def reset_driver_and_session(self):
|
|
self.tear_down_driver()
|
|
self.setup_driver_and_session()
|
|
|
|
def setup_driver_and_session(self):
|
|
self.display = driver_factory.virtual_display_if_enabled(headless_selenium())
|
|
self.driver = get_driver()
|
|
# New workflow index page does not degrade well to smaller sizes, needed
|
|
# to increase this.
|
|
self.driver.set_window_size(1280, 900)
|
|
|
|
if self.ensure_registered:
|
|
self.register()
|
|
|
|
def tear_down_driver(self):
|
|
exception = None
|
|
try:
|
|
self.driver.close()
|
|
except Exception as e:
|
|
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 default_timeout(self):
|
|
return DEFAULT_WAIT_TIMEOUT
|
|
|
|
def build_url(self, url, for_selenium=True):
|
|
if for_selenium:
|
|
base = self.target_url_from_selenium
|
|
else:
|
|
base = self.url
|
|
return urljoin(base, url)
|
|
|
|
@property
|
|
def test_data(self):
|
|
return self.navigation_data
|
|
|
|
def assert_initial_history_panel_state_correct(self):
|
|
# Move into a TestsHistoryPanel mixin
|
|
unnamed_name = self.test_data["historyPanel"]["text"]["history"]["newName"]
|
|
|
|
name_element = self.history_panel_name_element()
|
|
assert name_element.is_displayed()
|
|
assert unnamed_name in name_element.text
|
|
|
|
size_selector = self.test_data["historyPanel"]["selectors"]["history"]["size"]
|
|
initial_size_str = self.test_data["historyPanel"]["text"]["history"]["newSize"]
|
|
|
|
size_element = self.wait_for_selector(size_selector)
|
|
assert size_element.is_displayed()
|
|
assert initial_size_str in size_element.text, "%s not in %s" % (initial_size_str, size_element.text)
|
|
|
|
empty_msg_selector = self.test_data["historyPanel"]["selectors"]["history"]["emptyMsg"]
|
|
empty_msg_str = self.test_data["historyPanel"]["text"]["history"]["emptyMsg"]
|
|
|
|
empty_msg_element = self.wait_for_selector(empty_msg_selector)
|
|
assert empty_msg_element.is_displayed()
|
|
assert empty_msg_str in empty_msg_element.text
|
|
|
|
@property
|
|
def workflow_populator(self):
|
|
return SeleniumSessionWorkflowPopulator(self)
|
|
|
|
|
|
class SharedStateSeleniumTestCase(SeleniumTestCase):
|
|
"""This describes a class Selenium tests that setup class state for all tests.
|
|
|
|
This is a bit hacky because we are simulating class level initialization
|
|
with instance level methods. The problem is that super.setUp() works at
|
|
instance level. It might be worth considering having two variants of
|
|
SeleniumTestCase - one that initializes with the class and the other that
|
|
initializes with the instance but all the helpers are instance helpers.
|
|
"""
|
|
|
|
shared_state_initialized = False
|
|
shared_state_in_error = False
|
|
|
|
def setUp(self):
|
|
super(SharedStateSeleniumTestCase, self).setUp()
|
|
if not self.__class__.shared_state_initialized:
|
|
try:
|
|
self.setup_shared_state()
|
|
self.logout_if_needed()
|
|
except Exception:
|
|
self.__class__.shared_state_in_error = True
|
|
raise
|
|
finally:
|
|
self.__class__.shared_state_initialized = True
|
|
else:
|
|
if self.__class__.shared_state_in_error:
|
|
raise unittest.SkipTest("Skipping test, failed to initialize state previously.")
|
|
|
|
def setup_shared_state(self):
|
|
"""Override this to setup shared data for tests that gets initialized only once."""
|
|
|
|
|
|
class UsesHistoryItemAssertions:
|
|
|
|
def assert_item_peek_includes(self, hid, expected):
|
|
item_body_selector = self.history_panel_item_body_selector(hid=hid, wait=True)
|
|
peek_selector = item_body_selector + ' ' + self.test_data["historyPanel"]["selectors"]["hda"]["peek"]
|
|
peek_selector = self.wait_for_selector_visible(peek_selector)
|
|
|
|
def assert_item_info_includes(self, hid, expected):
|
|
item_body_selector = self.history_panel_item_body_selector(hid=hid, wait=True)
|
|
info_selector = item_body_selector + ' ' + self.test_data["historyPanel"]["selectors"]["hda"]["info"]
|
|
info_element = self.wait_for_selector_visible(info_selector)
|
|
text = info_element.text
|
|
assert expected in text, "Failed to find expected info text [%s] in info [%s]" % (expected, text)
|
|
|
|
def assert_item_dbkey_displayed_as(self, hid, dbkey):
|
|
item_body_selector = self.history_panel_item_body_selector(hid=hid, wait=True)
|
|
dbkey_selector = item_body_selector + ' ' + self.test_data["historyPanel"]["selectors"]["hda"]["dbkey"]
|
|
dbkey_element = self.wait_for_selector_visible(dbkey_selector)
|
|
assert dbkey in dbkey_element.text
|
|
|
|
def assert_item_summary_includes(self, hid, expected_text):
|
|
item_body_selector = self.history_panel_item_body_selector(hid=hid, wait=True)
|
|
summary_selector = "%s %s" % (item_body_selector, self.test_data["historyPanel"]["selectors"]["hda"]["summary"])
|
|
summary_element = self.wait_for_selector_visible(summary_selector)
|
|
text = summary_element.text
|
|
assert expected_text in text, "Expected summary [%s] not found in [%s]." % (expected_text, text)
|
|
|
|
def assert_item_name(self, hid, name):
|
|
item_selector = self.history_panel_item_selector(hid, wait=True)
|
|
title_selector = item_selector + ' ' + self.test_data["historyPanel"]["selectors"]["hda"]["name"]
|
|
title_element = self.wait_for_selector_visible(title_selector)
|
|
assert title_element.text == name, title_element.text
|
|
|
|
def assert_item_hid_text(self, hid):
|
|
# Check the text HID matches HID returned from API.
|
|
item_selector = self.history_panel_item_selector(hid, wait=True)
|
|
hid_selector = item_selector + ' ' + self.test_data["historyPanel"]["selectors"]["hda"]["hid"]
|
|
hid_element = self.wait_for_selector_visible(hid_selector)
|
|
assert hid_element.text == str(hid), hid_element.text
|
|
|
|
def _assert_item_button(self, buttons_area, expected_button, button_def):
|
|
selector = button_def["selector"]
|
|
# Let old tooltip expire, etc...
|
|
time.sleep(1)
|
|
button_item = self.wait_for_selector_visible("%s %s" % (buttons_area, selector))
|
|
expected_tooltip = button_def.get("tooltip")
|
|
self.assert_tooltip_text(button_item, expected_tooltip)
|
|
|
|
|
|
def default_web_host_for_selenium_tests():
|
|
if asbool(GALAXY_TEST_SELENIUM_REMOTE):
|
|
try:
|
|
dev_ip = get_ip_address('docker0')
|
|
return dev_ip
|
|
except IOError:
|
|
return DEFAULT_WEB_HOST
|
|
else:
|
|
return DEFAULT_WEB_HOST
|
|
|
|
|
|
def get_driver():
|
|
if asbool(GALAXY_TEST_SELENIUM_REMOTE):
|
|
return get_remote_driver()
|
|
else:
|
|
return get_local_driver()
|
|
|
|
|
|
def headless_selenium():
|
|
if asbool(GALAXY_TEST_SELENIUM_REMOTE):
|
|
return False
|
|
|
|
if GALAXY_TEST_SELENIUM_HEADLESS == "auto":
|
|
if driver_factory.is_virtual_display_available():
|
|
return True
|
|
else:
|
|
return False
|
|
else:
|
|
return asbool(GALAXY_TEST_SELENIUM_HEADLESS)
|
|
|
|
|
|
def get_local_driver():
|
|
return driver_factory.get_local_driver(GALAXY_TEST_SELENIUM_BROWSER)
|
|
|
|
|
|
def get_remote_driver():
|
|
return driver_factory.get_remote_driver(
|
|
host=GALAXY_TEST_SELENIUM_REMOTE_HOST,
|
|
port=GALAXY_TEST_SELENIUM_REMOTE_PORT,
|
|
browser=GALAXY_TEST_SELENIUM_BROWSER,
|
|
)
|
|
|
|
|
|
class SeleniumSessionGetPostMixin:
|
|
"""Mixin for adapting Galaxy testing populators helpers to Selenium session backed bioblend."""
|
|
|
|
def _get(self, route):
|
|
return self.selenium_test_case.api_get(route)
|
|
|
|
def _post(self, route, data={}):
|
|
full_url = self.selenium_test_case.build_url("api/" + route, for_selenium=False)
|
|
response = requests.post(full_url, data=data, cookies=self.selenium_test_case.selenium_to_requests_cookies())
|
|
return response
|
|
|
|
def __url(self, route):
|
|
return self._gi.url + "/" + route
|
|
|
|
|
|
class SeleniumSessionDatasetPopulator(populators.BaseDatasetPopulator, SeleniumSessionGetPostMixin):
|
|
|
|
"""Implementation of BaseDatasetPopulator backed by bioblend."""
|
|
|
|
def __init__(self, selenium_test_case):
|
|
"""Construct a dataset populator from a bioblend GalaxyInstance."""
|
|
self.selenium_test_case = selenium_test_case
|
|
|
|
|
|
class SeleniumSessionDatasetCollectionPopulator(populators.BaseDatasetCollectionPopulator, SeleniumSessionGetPostMixin):
|
|
|
|
"""Implementation of BaseDatasetCollectionPopulator backed by bioblend."""
|
|
|
|
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
|
|
|
|
|
|
class SeleniumSessionWorkflowPopulator(populators.BaseWorkflowPopulator, SeleniumSessionGetPostMixin, ImporterGalaxyInterface):
|
|
|
|
"""Implementation of BaseWorkflowPopulator backed by bioblend."""
|
|
|
|
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)
|
|
|
|
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()
|
|
|
|
def upload_yaml_workflow(self, has_yaml, **kwds):
|
|
workflow = convert_and_import_workflow(has_yaml, galaxy_interface=self, **kwds)
|
|
return workflow["id"]
|