mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-24 16:30:27 +08:00
Merge pull request #4536 from jmchilton/test_refactoring
Refactor test modules toward cleaner dependencies
This commit is contained in:
Björn Grüning
+1
-1
@@ -10,7 +10,7 @@ from .api_asserts import (
|
||||
)
|
||||
from .api_util import get_master_api_key, get_user_api_key
|
||||
from .interactor import GalaxyInteractorApi as BaseInteractor
|
||||
from .twilltestcase import FunctionalTestCase
|
||||
from .testcase import FunctionalTestCase
|
||||
|
||||
TEST_USER = "user@bx.psu.edu"
|
||||
ADMIN_TEST_USER = "test@bx.psu.edu"
|
||||
|
||||
@@ -21,7 +21,6 @@ import nose.loader
|
||||
import nose.plugins.manager
|
||||
from paste import httpserver
|
||||
|
||||
from functional import database_contexts
|
||||
from galaxy.app import UniverseApplication as GalaxyUniverseApplication
|
||||
from galaxy.util import asbool, download_to_file
|
||||
from galaxy.util.properties import load_app_properties
|
||||
@@ -54,6 +53,13 @@ DEFAULT_LOCALES = "en"
|
||||
log = logging.getLogger("test_driver")
|
||||
|
||||
|
||||
# Global variables to pass database contexts around - only needed for older
|
||||
# Tool Shed twill tests that didn't utilize the API for such interactions.
|
||||
galaxy_context = None
|
||||
tool_shed_context = None
|
||||
install_context = None
|
||||
|
||||
|
||||
def setup_tool_shed_tmp_dir():
|
||||
tool_shed_test_tmp_dir = os.environ.get('TOOL_SHED_TEST_TMP_DIR', None)
|
||||
if tool_shed_test_tmp_dir is None:
|
||||
@@ -480,8 +486,12 @@ def build_galaxy_app(simple_kwargs):
|
||||
# Build the Universe Application
|
||||
app = GalaxyUniverseApplication(**simple_kwargs)
|
||||
log.info("Embedded Galaxy application started")
|
||||
database_contexts.galaxy_context = app.model.context
|
||||
database_contexts.install_context = app.install_model.context
|
||||
|
||||
global galaxy_context
|
||||
global install_context
|
||||
galaxy_context = app.model.context
|
||||
install_context = app.install_model.context
|
||||
|
||||
return app
|
||||
|
||||
|
||||
@@ -497,8 +507,11 @@ def build_shed_app(simple_kwargs):
|
||||
simple_kwargs['global_conf'] = get_webapp_global_conf()
|
||||
|
||||
app = ToolshedUniverseApplication(**simple_kwargs)
|
||||
database_contexts.tool_shed_context = app.model.context
|
||||
log.info("Embedded Toolshed application started")
|
||||
|
||||
global tool_shed_context
|
||||
tool_shed_context = app.model.context
|
||||
|
||||
return app
|
||||
|
||||
|
||||
|
||||
@@ -1,224 +0,0 @@
|
||||
from sqlalchemy import and_, desc, false
|
||||
|
||||
import galaxy.model
|
||||
from functional import database_contexts
|
||||
|
||||
# Deprecated - import database_contexts and use galaxy_context
|
||||
sa_session = database_contexts.galaxy_context
|
||||
|
||||
|
||||
def gx_context():
|
||||
return database_contexts.galaxy_context
|
||||
|
||||
|
||||
def delete_obj(obj):
|
||||
gx_context().delete(obj)
|
||||
gx_context().flush()
|
||||
|
||||
|
||||
def delete_request_type_permissions(id):
|
||||
rtps = gx_context().query(galaxy.model.RequestTypePermissions) \
|
||||
.filter(and_(galaxy.model.RequestTypePermissions.table.c.request_type_id == id)) \
|
||||
.order_by(desc(galaxy.model.RequestTypePermissions.table.c.create_time))
|
||||
for rtp in rtps:
|
||||
gx_context().delete(rtp)
|
||||
gx_context().flush()
|
||||
|
||||
|
||||
def delete_user_roles(user):
|
||||
for ura in user.roles:
|
||||
gx_context().delete(ura)
|
||||
gx_context().flush()
|
||||
|
||||
|
||||
def flush(obj):
|
||||
gx_context().add(obj)
|
||||
gx_context().flush()
|
||||
|
||||
|
||||
def get_all_histories_for_user(user):
|
||||
return gx_context().query(galaxy.model.History) \
|
||||
.filter(and_(galaxy.model.History.table.c.user_id == user.id,
|
||||
galaxy.model.History.table.c.deleted == false())) \
|
||||
.all()
|
||||
|
||||
|
||||
def get_dataset_permissions_by_dataset(dataset):
|
||||
return gx_context().query(galaxy.model.DatasetPermissions) \
|
||||
.filter(galaxy.model.DatasetPermissions.table.c.dataset_id == dataset.id) \
|
||||
.all()
|
||||
|
||||
|
||||
def get_dataset_permissions_by_role(role):
|
||||
return gx_context().query(galaxy.model.DatasetPermissions) \
|
||||
.filter(galaxy.model.DatasetPermissions.table.c.role_id == role.id) \
|
||||
.first()
|
||||
|
||||
|
||||
def get_default_history_permissions_by_history(history):
|
||||
return gx_context().query(galaxy.model.DefaultHistoryPermissions) \
|
||||
.filter(galaxy.model.DefaultHistoryPermissions.table.c.history_id == history.id) \
|
||||
.all()
|
||||
|
||||
|
||||
def get_default_history_permissions_by_role(role):
|
||||
return gx_context().query(galaxy.model.DefaultHistoryPermissions) \
|
||||
.filter(galaxy.model.DefaultHistoryPermissions.table.c.role_id == role.id) \
|
||||
.all()
|
||||
|
||||
|
||||
def get_default_user_permissions_by_role(role):
|
||||
return gx_context().query(galaxy.model.DefaultUserPermissions) \
|
||||
.filter(galaxy.model.DefaultUserPermissions.table.c.role_id == role.id) \
|
||||
.all()
|
||||
|
||||
|
||||
def get_default_user_permissions_by_user(user):
|
||||
return gx_context().query(galaxy.model.DefaultUserPermissions) \
|
||||
.filter(galaxy.model.DefaultUserPermissions.table.c.user_id == user.id) \
|
||||
.all()
|
||||
|
||||
|
||||
def get_form(name):
|
||||
fdc_list = gx_context().query(galaxy.model.FormDefinitionCurrent) \
|
||||
.filter(galaxy.model.FormDefinitionCurrent.table.c.deleted == false()) \
|
||||
.order_by(galaxy.model.FormDefinitionCurrent.table.c.create_time.desc())
|
||||
for fdc in fdc_list:
|
||||
gx_context().refresh(fdc)
|
||||
gx_context().refresh(fdc.latest_form)
|
||||
if fdc.latest_form.name == name:
|
||||
return fdc.latest_form
|
||||
return None
|
||||
|
||||
|
||||
def get_folder(parent_id, name, description):
|
||||
return gx_context().query(galaxy.model.LibraryFolder) \
|
||||
.filter(and_(galaxy.model.LibraryFolder.table.c.parent_id == parent_id,
|
||||
galaxy.model.LibraryFolder.table.c.name == name,
|
||||
galaxy.model.LibraryFolder.table.c.description == description)) \
|
||||
.first()
|
||||
|
||||
|
||||
def get_group_by_name(name):
|
||||
return gx_context().query(galaxy.model.Group).filter(galaxy.model.Group.table.c.name == name).first()
|
||||
|
||||
|
||||
def get_group_role_associations_by_group(group):
|
||||
return gx_context().query(galaxy.model.GroupRoleAssociation) \
|
||||
.filter(galaxy.model.GroupRoleAssociation.table.c.group_id == group.id) \
|
||||
.all()
|
||||
|
||||
|
||||
def get_group_role_associations_by_role(role):
|
||||
return gx_context().query(galaxy.model.GroupRoleAssociation) \
|
||||
.filter(galaxy.model.GroupRoleAssociation.table.c.role_id == role.id) \
|
||||
.all()
|
||||
|
||||
|
||||
def get_latest_dataset():
|
||||
return gx_context().query(galaxy.model.Dataset) \
|
||||
.order_by(desc(galaxy.model.Dataset.table.c.create_time)) \
|
||||
.first()
|
||||
|
||||
|
||||
def get_latest_hda():
|
||||
return gx_context().query(galaxy.model.HistoryDatasetAssociation) \
|
||||
.order_by(desc(galaxy.model.HistoryDatasetAssociation.table.c.create_time)) \
|
||||
.first()
|
||||
|
||||
|
||||
def get_latest_history_for_user(user):
|
||||
return gx_context().query(galaxy.model.History) \
|
||||
.filter(and_(galaxy.model.History.table.c.deleted == false(),
|
||||
galaxy.model.History.table.c.user_id == user.id)) \
|
||||
.order_by(desc(galaxy.model.History.table.c.create_time)) \
|
||||
.first()
|
||||
|
||||
|
||||
def get_latest_ldda_by_name(name):
|
||||
return gx_context().query(galaxy.model.LibraryDatasetDatasetAssociation) \
|
||||
.filter(and_(galaxy.model.LibraryDatasetDatasetAssociation.table.c.name == name,
|
||||
galaxy.model.LibraryDatasetDatasetAssociation.table.c.deleted == false())) \
|
||||
.order_by(desc(galaxy.model.LibraryDatasetDatasetAssociation.table.c.create_time)) \
|
||||
.first()
|
||||
|
||||
|
||||
def get_latest_lddas(limit):
|
||||
return gx_context().query(galaxy.model.LibraryDatasetDatasetAssociation) \
|
||||
.order_by(desc(galaxy.model.LibraryDatasetDatasetAssociation.table.c.update_time)) \
|
||||
.limit(limit)
|
||||
|
||||
|
||||
def get_library(name, description, synopsis):
|
||||
return gx_context().query(galaxy.model.Library) \
|
||||
.filter(and_(galaxy.model.Library.table.c.name == name,
|
||||
galaxy.model.Library.table.c.description == description,
|
||||
galaxy.model.Library.table.c.synopsis == synopsis,
|
||||
galaxy.model.Library.table.c.deleted == false())) \
|
||||
.first()
|
||||
|
||||
|
||||
def get_private_role(user):
|
||||
for role in user.all_roles():
|
||||
if role.name == user.email and role.description == 'Private Role for %s' % user.email:
|
||||
return role
|
||||
raise AssertionError("Private role not found for user '%s'" % user.email)
|
||||
|
||||
|
||||
def get_request_by_name(name):
|
||||
return gx_context().query(galaxy.model.Request) \
|
||||
.filter(and_(galaxy.model.Request.table.c.name == name,
|
||||
galaxy.model.Request.table.c.deleted == false())) \
|
||||
.first()
|
||||
|
||||
|
||||
def get_request_type_by_name(name):
|
||||
return gx_context().query(galaxy.model.RequestType) \
|
||||
.filter(and_(galaxy.model.RequestType.table.c.name == name)) \
|
||||
.order_by(desc(galaxy.model.RequestType.table.c.create_time)) \
|
||||
.first()
|
||||
|
||||
|
||||
def get_role_by_name(name):
|
||||
return gx_context().query(galaxy.model.Role).filter(galaxy.model.Role.table.c.name == name).first()
|
||||
|
||||
|
||||
def get_user(email):
|
||||
return gx_context().query(galaxy.model.User) \
|
||||
.filter(galaxy.model.User.table.c.email == email) \
|
||||
.first()
|
||||
|
||||
|
||||
def get_user_address(user, short_desc):
|
||||
return gx_context().query(galaxy.model.UserAddress) \
|
||||
.filter(and_(galaxy.model.UserAddress.table.c.user_id == user.id,
|
||||
galaxy.model.UserAddress.table.c.desc == short_desc,
|
||||
galaxy.model.UserAddress.table.c.deleted == false())) \
|
||||
.order_by(desc(galaxy.model.UserAddress.table.c.create_time)) \
|
||||
.first()
|
||||
|
||||
|
||||
def get_user_group_associations_by_group(group):
|
||||
return gx_context().query(galaxy.model.UserGroupAssociation) \
|
||||
.filter(galaxy.model.UserGroupAssociation.table.c.group_id == group.id) \
|
||||
.all()
|
||||
|
||||
|
||||
def get_user_info_form_definition():
|
||||
return galaxy.model.FormDefinition.types.USER_INFO
|
||||
|
||||
|
||||
def get_user_role_associations_by_role(role):
|
||||
return gx_context().query(galaxy.model.UserRoleAssociation) \
|
||||
.filter(galaxy.model.UserRoleAssociation.table.c.role_id == role.id) \
|
||||
.all()
|
||||
|
||||
|
||||
def mark_obj_deleted(obj):
|
||||
obj.deleted = True
|
||||
gx_context().add(obj)
|
||||
gx_context().flush()
|
||||
|
||||
|
||||
def refresh(obj):
|
||||
gx_context().refresh(obj)
|
||||
@@ -0,0 +1,148 @@
|
||||
from __future__ import print_function
|
||||
|
||||
import logging
|
||||
import os
|
||||
import time
|
||||
import unittest
|
||||
from json import loads
|
||||
|
||||
from galaxy.tools.verify import verify
|
||||
from galaxy.tools.verify.test_data import TestDataResolver
|
||||
from galaxy.web import security
|
||||
|
||||
from .driver_util import GalaxyTestDriver
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
DEFAULT_TOOL_TEST_WAIT = os.environ.get("GALAXY_TEST_DEFAULT_WAIT", 86400)
|
||||
|
||||
|
||||
class FunctionalTestCase(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
# Security helper
|
||||
self.security = security.SecurityHelper(id_secret='changethisinproductiontoo')
|
||||
self.history_id = os.environ.get('GALAXY_TEST_HISTORY_ID', None)
|
||||
self.host = os.environ.get('GALAXY_TEST_HOST')
|
||||
self.port = os.environ.get('GALAXY_TEST_PORT')
|
||||
default_url = "http://%s:%s" % (self.host, self.port)
|
||||
self.url = os.environ.get('GALAXY_TEST_EXTERNAL', default_url)
|
||||
self.test_data_resolver = TestDataResolver()
|
||||
tool_shed_test_file = os.environ.get('GALAXY_TOOL_SHED_TEST_FILE', None)
|
||||
if tool_shed_test_file:
|
||||
f = open(tool_shed_test_file, 'r')
|
||||
text = f.read()
|
||||
f.close()
|
||||
self.shed_tools_dict = loads(text)
|
||||
else:
|
||||
self.shed_tools_dict = {}
|
||||
self.keepOutdir = os.environ.get('GALAXY_TEST_SAVE', '')
|
||||
if self.keepOutdir > '':
|
||||
try:
|
||||
os.makedirs(self.keepOutdir)
|
||||
except:
|
||||
pass
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
"""Configure and start Galaxy for a test."""
|
||||
cls._test_driver = None
|
||||
|
||||
if not os.environ.get("GALAXY_TEST_ENVIRONMENT_CONFIGURED"):
|
||||
cls._test_driver = GalaxyTestDriver()
|
||||
cls._test_driver.setup(config_object=cls)
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
"""Shutdown Galaxy server and cleanup temp directory."""
|
||||
if cls._test_driver:
|
||||
cls._test_driver.tear_down()
|
||||
|
||||
def get_filename(self, filename, shed_tool_id=None):
|
||||
# For tool tests override get_filename to point at an installed tool if shed_tool_id is set.
|
||||
if shed_tool_id and getattr(self, "shed_tools_dict", None):
|
||||
file_dir = self.shed_tools_dict[shed_tool_id]
|
||||
if file_dir:
|
||||
return os.path.abspath(os.path.join(file_dir, filename))
|
||||
return self.test_data_resolver.get_filename(filename)
|
||||
|
||||
# TODO: Make this more generic, shouldn't be related to tool stuff I guess.
|
||||
def wait_for(self, func, **kwd):
|
||||
sleep_amount = 0.2
|
||||
slept = 0
|
||||
walltime_exceeded = kwd.get("maxseconds", None)
|
||||
if walltime_exceeded is None:
|
||||
walltime_exceeded = DEFAULT_TOOL_TEST_WAIT
|
||||
|
||||
exceeded = True
|
||||
while slept <= walltime_exceeded:
|
||||
result = func()
|
||||
if result:
|
||||
time.sleep(sleep_amount)
|
||||
slept += sleep_amount
|
||||
sleep_amount *= 2
|
||||
else:
|
||||
exceeded = False
|
||||
break
|
||||
|
||||
if exceeded:
|
||||
message = 'Tool test run exceeded walltime [total %s, max %s], terminating.' % (slept, walltime_exceeded)
|
||||
log.info(message)
|
||||
raise AssertionError(message)
|
||||
|
||||
# TODO: Move verify_xxx into GalaxyInteractor or some relevant mixin.
|
||||
def verify_hid(self, filename, hda_id, attributes, shed_tool_id, hid="", dataset_fetcher=None):
|
||||
assert dataset_fetcher is not None
|
||||
|
||||
def get_filename(test_filename):
|
||||
return self.get_filename(test_filename, shed_tool_id=shed_tool_id)
|
||||
|
||||
def verify_extra_files(extra_files):
|
||||
self._verify_extra_files_content(extra_files, hda_id, shed_tool_id=shed_tool_id, dataset_fetcher=dataset_fetcher)
|
||||
|
||||
data = dataset_fetcher(hda_id)
|
||||
item_label = "History item %s" % hid
|
||||
verify(
|
||||
item_label,
|
||||
data,
|
||||
attributes=attributes,
|
||||
filename=filename,
|
||||
get_filename=get_filename,
|
||||
keep_outputs_dir=self.keepOutdir,
|
||||
verify_extra_files=verify_extra_files,
|
||||
)
|
||||
|
||||
def _verify_composite_datatype_file_content(self, file_name, hda_id, base_name=None, attributes=None, dataset_fetcher=None, shed_tool_id=None):
|
||||
assert dataset_fetcher is not None
|
||||
|
||||
def get_filename(test_filename):
|
||||
return self.get_filename(test_filename, shed_tool_id=shed_tool_id)
|
||||
|
||||
data = dataset_fetcher(hda_id, base_name)
|
||||
item_label = "History item %s" % hda_id
|
||||
try:
|
||||
verify(
|
||||
item_label,
|
||||
data,
|
||||
attributes=attributes,
|
||||
filename=file_name,
|
||||
get_filename=get_filename,
|
||||
keep_outputs_dir=self.keepOutdir,
|
||||
)
|
||||
except AssertionError as err:
|
||||
errmsg = 'Composite file (%s) of %s different than expected, difference:\n' % (base_name, item_label)
|
||||
errmsg += str(err)
|
||||
raise AssertionError(errmsg)
|
||||
|
||||
def _verify_extra_files_content(self, extra_files, hda_id, dataset_fetcher, shed_tool_id=None):
|
||||
files_list = []
|
||||
for extra_type, extra_value, extra_name, extra_attributes in extra_files:
|
||||
if extra_type == 'file':
|
||||
files_list.append((extra_name, extra_value, extra_attributes))
|
||||
elif extra_type == 'directory':
|
||||
for filename in os.listdir(self.get_filename(extra_value, shed_tool_id=shed_tool_id)):
|
||||
files_list.append((filename, os.path.join(extra_value, filename), extra_attributes))
|
||||
else:
|
||||
raise ValueError('unknown extra_files type: %s' % extra_type)
|
||||
for filename, filepath, attributes in files_list:
|
||||
self._verify_composite_datatype_file_content(filepath, hda_id, base_name=filename, attributes=attributes, dataset_fetcher=dataset_fetcher, shed_tool_id=shed_tool_id)
|
||||
@@ -1,11 +1,7 @@
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
|
||||
galaxy_root = os.path.abspath(os.path.join(os.path.dirname(__file__), os.path.pardir, os.path.pardir))
|
||||
sys.path.insert(1, os.path.join(galaxy_root, 'lib'))
|
||||
|
||||
from galaxy.util import parse_xml
|
||||
from galaxy.util import galaxy_root_path, parse_xml
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
@@ -64,7 +60,7 @@ def parse_tool_panel_config(config, shed_tools_dict):
|
||||
has_test_data = True
|
||||
if galaxy_test_file_dir != last_galaxy_test_file_dir:
|
||||
if not os.path.isabs(galaxy_test_file_dir):
|
||||
galaxy_test_file_dir = os.path.join(galaxy_root, galaxy_test_file_dir)
|
||||
galaxy_test_file_dir = os.path.join(galaxy_root_path, galaxy_test_file_dir)
|
||||
guid = elem.get('guid')
|
||||
shed_tools_dict[guid] = galaxy_test_file_dir
|
||||
last_galaxy_test_file_dir = galaxy_test_file_dir
|
||||
@@ -83,7 +79,7 @@ def parse_tool_panel_config(config, shed_tools_dict):
|
||||
has_test_data = True
|
||||
if galaxy_test_file_dir != last_galaxy_test_file_dir:
|
||||
if not os.path.isabs(galaxy_test_file_dir):
|
||||
galaxy_test_file_dir = os.path.join(galaxy_root, galaxy_test_file_dir)
|
||||
galaxy_test_file_dir = os.path.join(galaxy_root_path, galaxy_test_file_dir)
|
||||
guid = section_elem.get('guid')
|
||||
shed_tools_dict[guid] = galaxy_test_file_dir
|
||||
last_galaxy_test_file_dir = galaxy_test_file_dir
|
||||
|
||||
@@ -1,6 +0,0 @@
|
||||
# Global variables to pass database contexts around. Fairly hackish that they
|
||||
# are shared this way, but at least they have been moved out of Galaxy's lib/
|
||||
# code base.
|
||||
galaxy_context = None
|
||||
tool_shed_context = None
|
||||
install_context = None
|
||||
File diff suppressed because it is too large
Load Diff
@@ -12,11 +12,12 @@ except ImportError:
|
||||
|
||||
from base.instrument import register_job_data
|
||||
from base.interactor import build_interactor, RunToolException, stage_data_in_history
|
||||
from base.twilltestcase import TwillTestCase
|
||||
from galaxy.tools import DataManagerTool
|
||||
from galaxy.tools.verify.asserts import verify_assertions
|
||||
from galaxy.util import bunch
|
||||
|
||||
from .twilltestcase import TwillTestCase
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
toolbox = None
|
||||
@@ -26,7 +27,11 @@ TOOL_TYPES_NO_TEST = (DataManagerTool, )
|
||||
|
||||
|
||||
class ToolTestCase(TwillTestCase):
|
||||
"""Abstract test case that runs tests based on a `galaxy.tools.test.ToolTest`"""
|
||||
"""Abstract test case that runs tests based on a `galaxy.tools.test.ToolTest`.
|
||||
|
||||
Ideally this would be FunctionalTestCase instead of a TwillTestCase but the
|
||||
subclass DataManagerToolTestCase requires the use of Twill still.
|
||||
"""
|
||||
|
||||
def do_it(self, testdef, resource_parameters={}):
|
||||
"""
|
||||
@@ -255,6 +260,16 @@ class ToolTestCase(TwillTestCase):
|
||||
else:
|
||||
return job_stdio
|
||||
|
||||
def _format_stream(self, output, stream, format):
|
||||
output = output or ''
|
||||
if format:
|
||||
msg = "---------------------- >> begin tool %s << -----------------------\n" % stream
|
||||
msg += output + "\n"
|
||||
msg += "----------------------- >> end tool %s << ------------------------\n" % stream
|
||||
else:
|
||||
msg = output
|
||||
return msg
|
||||
|
||||
|
||||
class JobOutputsError(AssertionError):
|
||||
|
||||
|
||||
@@ -3,8 +3,6 @@ from __future__ import print_function
|
||||
import logging
|
||||
import os
|
||||
import tempfile
|
||||
import time
|
||||
import unittest
|
||||
from json import loads
|
||||
from xml.etree import ElementTree
|
||||
|
||||
@@ -17,11 +15,7 @@ from six import string_types, StringIO
|
||||
from six.moves.urllib.parse import urlencode, urlparse
|
||||
from twill.other_packages._mechanize_dist import ClientForm
|
||||
|
||||
from galaxy.tools.verify import verify
|
||||
from galaxy.tools.verify.test_data import TestDataResolver
|
||||
from galaxy.web import security
|
||||
|
||||
from .driver_util import GalaxyTestDriver
|
||||
from base.testcase import FunctionalTestCase
|
||||
|
||||
# Force twill to log to a buffer -- FIXME: Should this go to stdout and be captured by nose?
|
||||
buffer = StringIO()
|
||||
@@ -35,137 +29,6 @@ log = logging.getLogger(__name__)
|
||||
DEFAULT_TOOL_TEST_WAIT = os.environ.get("GALAXY_TEST_DEFAULT_WAIT", 86400)
|
||||
|
||||
|
||||
class FunctionalTestCase(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
# Security helper
|
||||
self.security = security.SecurityHelper(id_secret='changethisinproductiontoo')
|
||||
self.history_id = os.environ.get('GALAXY_TEST_HISTORY_ID', None)
|
||||
self.host = os.environ.get('GALAXY_TEST_HOST')
|
||||
self.port = os.environ.get('GALAXY_TEST_PORT')
|
||||
default_url = "http://%s:%s" % (self.host, self.port)
|
||||
self.url = os.environ.get('GALAXY_TEST_EXTERNAL', default_url)
|
||||
self.test_data_resolver = TestDataResolver()
|
||||
tool_shed_test_file = os.environ.get('GALAXY_TOOL_SHED_TEST_FILE', None)
|
||||
if tool_shed_test_file:
|
||||
f = open(tool_shed_test_file, 'r')
|
||||
text = f.read()
|
||||
f.close()
|
||||
self.shed_tools_dict = loads(text)
|
||||
else:
|
||||
self.shed_tools_dict = {}
|
||||
self.keepOutdir = os.environ.get('GALAXY_TEST_SAVE', '')
|
||||
if self.keepOutdir > '':
|
||||
try:
|
||||
os.makedirs(self.keepOutdir)
|
||||
except:
|
||||
pass
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
"""Configure and start Galaxy for a test."""
|
||||
cls._test_driver = None
|
||||
|
||||
if not os.environ.get("GALAXY_TEST_ENVIRONMENT_CONFIGURED"):
|
||||
cls._test_driver = GalaxyTestDriver()
|
||||
cls._test_driver.setup(config_object=cls)
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
"""Shutdown Galaxy server and cleanup temp directory."""
|
||||
if cls._test_driver:
|
||||
cls._test_driver.tear_down()
|
||||
|
||||
def get_filename(self, filename, shed_tool_id=None):
|
||||
# For tool tests override get_filename to point at an installed tool if shed_tool_id is set.
|
||||
if shed_tool_id and getattr(self, "shed_tools_dict", None):
|
||||
file_dir = self.shed_tools_dict[shed_tool_id]
|
||||
if file_dir:
|
||||
return os.path.abspath(os.path.join(file_dir, filename))
|
||||
return self.test_data_resolver.get_filename(filename)
|
||||
|
||||
# TODO: Make this more generic, shouldn't be related to tool stuff I guess.
|
||||
def wait_for(self, func, **kwd):
|
||||
sleep_amount = 0.2
|
||||
slept = 0
|
||||
walltime_exceeded = kwd.get("maxseconds", None)
|
||||
if walltime_exceeded is None:
|
||||
walltime_exceeded = DEFAULT_TOOL_TEST_WAIT
|
||||
|
||||
exceeded = True
|
||||
while slept <= walltime_exceeded:
|
||||
result = func()
|
||||
if result:
|
||||
time.sleep(sleep_amount)
|
||||
slept += sleep_amount
|
||||
sleep_amount *= 2
|
||||
else:
|
||||
exceeded = False
|
||||
break
|
||||
|
||||
if exceeded:
|
||||
message = 'Tool test run exceeded walltime [total %s, max %s], terminating.' % (slept, walltime_exceeded)
|
||||
log.info(message)
|
||||
raise AssertionError(message)
|
||||
|
||||
# TODO: Move verify_xxx into GalaxyInteractor or some relevant mixin.
|
||||
def verify_hid(self, filename, hda_id, attributes, shed_tool_id, hid="", dataset_fetcher=None):
|
||||
assert dataset_fetcher is not None
|
||||
|
||||
def get_filename(test_filename):
|
||||
return self.get_filename(test_filename, shed_tool_id=shed_tool_id)
|
||||
|
||||
def verify_extra_files(extra_files):
|
||||
self._verify_extra_files_content(extra_files, hda_id, shed_tool_id=shed_tool_id, dataset_fetcher=dataset_fetcher)
|
||||
|
||||
data = dataset_fetcher(hda_id)
|
||||
item_label = "History item %s" % hid
|
||||
verify(
|
||||
item_label,
|
||||
data,
|
||||
attributes=attributes,
|
||||
filename=filename,
|
||||
get_filename=get_filename,
|
||||
keep_outputs_dir=self.keepOutdir,
|
||||
verify_extra_files=verify_extra_files,
|
||||
)
|
||||
|
||||
def _verify_composite_datatype_file_content(self, file_name, hda_id, base_name=None, attributes=None, dataset_fetcher=None, shed_tool_id=None):
|
||||
assert dataset_fetcher is not None
|
||||
|
||||
def get_filename(test_filename):
|
||||
return self.get_filename(test_filename, shed_tool_id=shed_tool_id)
|
||||
|
||||
data = dataset_fetcher(hda_id, base_name)
|
||||
item_label = "History item %s" % hda_id
|
||||
try:
|
||||
verify(
|
||||
item_label,
|
||||
data,
|
||||
attributes=attributes,
|
||||
filename=file_name,
|
||||
get_filename=get_filename,
|
||||
keep_outputs_dir=self.keepOutdir,
|
||||
)
|
||||
except AssertionError as err:
|
||||
errmsg = 'Composite file (%s) of %s different than expected, difference:\n' % (base_name, item_label)
|
||||
errmsg += str(err)
|
||||
raise AssertionError(errmsg)
|
||||
|
||||
def _verify_extra_files_content(self, extra_files, hda_id, dataset_fetcher, shed_tool_id=None):
|
||||
files_list = []
|
||||
for extra_type, extra_value, extra_name, extra_attributes in extra_files:
|
||||
if extra_type == 'file':
|
||||
files_list.append((extra_name, extra_value, extra_attributes))
|
||||
elif extra_type == 'directory':
|
||||
for filename in os.listdir(self.get_filename(extra_value, shed_tool_id=shed_tool_id)):
|
||||
files_list.append((filename, os.path.join(extra_value, filename), extra_attributes))
|
||||
else:
|
||||
raise ValueError('unknown extra_files type: %s' % extra_type)
|
||||
for filename, filepath, attributes in files_list:
|
||||
self._verify_composite_datatype_file_content(filepath, hda_id, base_name=filename, attributes=attributes, dataset_fetcher=dataset_fetcher, shed_tool_id=shed_tool_id)
|
||||
|
||||
|
||||
class TwillTestCase(FunctionalTestCase):
|
||||
|
||||
"""Class of FunctionalTestCase geared toward HTML interactions using the Twill library."""
|
||||
@@ -566,13 +429,3 @@ class TwillTestCase(FunctionalTestCase):
|
||||
f.write(content)
|
||||
f.close()
|
||||
return fname
|
||||
|
||||
def _format_stream(self, output, stream, format):
|
||||
output = output or ''
|
||||
if format:
|
||||
msg = "---------------------- >> begin tool %s << -----------------------\n" % stream
|
||||
msg += output + "\n"
|
||||
msg += "----------------------- >> end tool %s << ------------------------\n" % stream
|
||||
else:
|
||||
msg = output
|
||||
return msg
|
||||
@@ -27,7 +27,7 @@ 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.testcase import FunctionalTestCase
|
||||
from base.workflows_format_2 import (
|
||||
ImporterGalaxyInterface,
|
||||
convert_and_import_workflow,
|
||||
|
||||
@@ -6,7 +6,7 @@ import galaxy.model
|
||||
import galaxy.model.tool_shed_install
|
||||
import galaxy.webapps.tool_shed.model as model
|
||||
|
||||
from functional.database_contexts import (galaxy_context as ga_session,
|
||||
from base.driver_util import (galaxy_context as ga_session,
|
||||
install_context as install_session, tool_shed_context as sa_session)
|
||||
|
||||
log = logging.getLogger('test.tool_shed.test_db_util')
|
||||
|
||||
@@ -16,7 +16,7 @@ import galaxy.model.tool_shed_install as galaxy_model
|
||||
import galaxy.util
|
||||
import galaxy.webapps.tool_shed.util.hgweb_config
|
||||
from base.tool_shed_util import repository_installation_timeout
|
||||
from base.twilltestcase import TwillTestCase
|
||||
from functional.twilltestcase import TwillTestCase
|
||||
from galaxy.web import security
|
||||
from tool_shed.util import hg_util, xml_util
|
||||
from tool_shed.util.encoding_util import tool_shed_encode
|
||||
|
||||
Reference in New Issue
Block a user