galaxy/test/selenium_tests/framework.py

"""Basis for Selenium test framework."""
from __future__ import absolute_import
from __future__ import print_function

import datetime
import json
import os
import traceback
import unittest
from functools import partial, wraps

import requests
try:
    from pyvirtualdisplay import Display
except ImportError:
    Display = None
from six.moves.urllib.parse import urljoin

from base import populators  # noqa: I100
from base.driver_util import classproperty, DEFAULT_WEB_HOST, get_ip_address  # noqa: I100
from base.testcase import FunctionalTestCase  # noqa: I100
from base.workflows_format_2 import (  # noqa: I100
    convert_and_import_workflow,
    ImporterGalaxyInterface,
)
from galaxy_selenium import (  # noqa: I100
    driver_factory,
)
from galaxy_selenium.navigates_galaxy import NavigatesGalaxy, retry_during_transitions  # noqa: I100
from galaxy.util import asbool

DEFAULT_TIMEOUT_MULTIPLIER = 1
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"

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)
# 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"))

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)


try:
    from nose.tools import nottest
except ImportError:
    def nottest(x):
        return x


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:
                current_history_id = self.current_history_id()
                self.api_delete("histories/%s" % current_history_id)

    return func_wrapper


@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, 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)
                    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 snapshot in getattr(self, "snapshots", []):
                        snapshot.write_to_error_directory(write_file)

                    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 TestSnapshot(object):

    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()

    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))


class SeleniumTestCase(FunctionalTestCase, NavigatesGalaxy):
    # 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

    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.snapshots = []
        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 snapshot(self, description):
        self.snapshots.append(TestSnapshot(self.driver, len(self.snapshots), description))

    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.login()

    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()

    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 timeout_multiplier(self):
        return TIMEOUT_MULTIPLIER

    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...
        self.sleep_for(self.wait_types.UX_TRANSITION)
        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"]