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 time

import traceback

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.twilltestcase 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 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"]