opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-09-24 16:30:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

Serialize tool test descriptions through the API.

Browse Source 
Use this is in tool testing. Prevents needing to load / have tools in memory to run tool tests - should allow external tool testing.
This commit is contained in:
John Chilton
2018-03-06 20:07:51 -05:00
parent 508047f00e
commit c3af6c8fea
10 changed files with 348 additions and 150 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/tools/parser/interface.py
+75 -7
View File
@@ -357,21 +357,74 @@ class ToolStdioExitCode(object):
class TestCollectionDef(object):
# TODO: do not require XML directly here.
def __init__(self, elem, parse_param_elem):
self.elements = []
def __init__(self, attrib, name, collection_type, elements):
self.attrib = attrib
self.collection_type = collection_type
self.elements = elements
self.name = name
@staticmethod
def from_xml(elem, parse_param_elem):
elements = []
attrib = dict(elem.attrib)
self.collection_type = attrib["type"]
self.name = attrib.get("name", "Unnamed Collection")
collection_type = attrib["type"]
name = attrib.get("name", "Unnamed Collection")
for element in elem.findall("element"):
element_attrib = dict(element.attrib)
element_identifier = element_attrib["name"]
nested_collection_elem = element.find("collection")
if nested_collection_elem is not None:
self.elements.append((element_identifier, TestCollectionDef(nested_collection_elem, parse_param_elem)))
elements.append((element_identifier, TestCollectionDef.from_xml(nested_collection_elem, parse_param_elem)))
else:
self.elements.append((element_identifier, parse_param_elem(element)))
elements.append((element_identifier, parse_param_elem(element)))
return TestCollectionDef(
attrib=attrib,
collection_type=collection_type,
elements=elements,
name=name,
)
def to_dict(self):
def element_to_dict(element_pair):
element_identifier, element_def = element_pair
if isinstance(element_def, TestCollectionDef):
element_def = element_def.to_dict()
return {
"element_identifier": element_identifier,
"element_def": element_def,
}
return {
"model_class": "TestCollectionDef",
"attrib": self.attrib,
"collection_type": self.collection_type,
"elements": map(element_to_dict, self.elements or []),
"name": self.name,
}
@staticmethod
def from_dict(as_dict):
assert as_dict["model_class"] == "TestCollectionDef"
def element_from_dict(element_dict):
if "element_def" not in element_dict:
raise Exception("Invalid element_dict %s" % element_dict)
element_def = element_dict["element_def"]
# TODO: stop using tuples and use dicts internally to eliminate this check
if not isinstance(element_def, dict):
pass
elif element_def.get("model_class", None) == "TestCollectionDef":
element_def = TestCollectionDef.from_dict(element_def)
return (element_dict["element_identifier"], element_def)
return TestCollectionDef(
attrib=as_dict["attrib"],
name=as_dict["name"],
elements=map(element_from_dict, as_dict["elements"] or []),
collection_type=as_dict["collection_type"],
)
def collect_inputs(self):
inputs = []
@@ -393,3 +446,18 @@ class TestCollectionOutputDef(object):
self.count = int(count) if count is not None else None
self.attrib = attrib
self.element_tests = element_tests
@staticmethod
def from_dict(as_dict):
return TestCollectionOutputDef(
name=as_dict["name"],
attrib=as_dict["attrib"],
element_tests=as_dict["element_tests"],
)
def to_dict(self):
return dict(
name=self.name,
attrib=self.attrib,
element_tests=self.element_tests
)
lib/galaxy/tools/parser/xml.py
+10 -13
View File
@@ -391,8 +391,6 @@ class XmlToolSource(ToolSource):
for i, test_elem in enumerate(tests_elem.findall("test")):
tests.append(_test_elem_to_dict(test_elem, i))
_copy_to_dict_if_present(tests_elem, rval, ["interactor"])
return rval
def parse_profile(self):
@@ -431,7 +429,7 @@ def __parse_output_elems(test_elem):
outputs = []
for output_elem in test_elem.findall("output"):
name, file, attributes = __parse_output_elem(output_elem)
outputs.append((name, file, attributes))
outputs.append([name, file, attributes])
return outputs
@@ -464,7 +462,7 @@ def __parse_output_collection_elem(output_collection_elem):
if name is None:
raise Exception("Test output collection does not have a 'name'")
element_tests = __parse_element_tests(output_collection_elem)
return TestCollectionOutputDef(name, attrib, element_tests)
return TestCollectionOutputDef(name, attrib, element_tests).to_dict()
def __parse_element_tests(parent_element):
@@ -641,33 +639,32 @@ def __parse_param_elem(param_elem, i=0):
value = attrib['value']
else:
value = None
attrib['children'] = param_elem
if attrib['children'] is not None:
children_elem = param_elem
if children_elem is not None:
# At this time, we can assume having children only
# occurs on DataToolParameter test items but this could
# change and would cause the below parsing to change
# based upon differences in children items
attrib['metadata'] = []
attrib['metadata'] = {}
attrib['composite_data'] = []
attrib['edit_attributes'] = []
# Composite datasets need to be renamed uniquely
composite_data_name = None
for child in attrib['children']:
for child in children_elem:
if child.tag == 'composite_data':
attrib['composite_data'].append(child)
file_name = child.get("value")
attrib['composite_data'].append(file_name)
if composite_data_name is None:
# Generate a unique name; each test uses a
# fresh history.
composite_data_name = '_COMPOSITE_RENAMED_t%d_%s' \
% (i, uuid.uuid1().hex)
elif child.tag == 'metadata':
attrib['metadata'].append(child)
elif child.tag == 'metadata':
attrib['metadata'].append(child)
attrib['metadata'][child.get("name")] = child.get("value")
elif child.tag == 'edit_attributes':
attrib['edit_attributes'].append(child)
elif child.tag == 'collection':
attrib['collection'] = TestCollectionDef(child, __parse_param_elem)
attrib['collection'] = TestCollectionDef.from_xml(child, __parse_param_elem)
if composite_data_name:
# Composite datasets need implicit renaming;
# inserted at front of list so explicit declarations
lib/galaxy/tools/test.py
+4 -89
View File
@@ -6,6 +6,7 @@ from six import string_types
import galaxy.tools.parameters.basic
import galaxy.tools.parameters.grouping
from galaxy.tools.verify.interactor import ToolTestDescription
from galaxy.util import string_as_bool
try:
@@ -16,11 +17,6 @@ except ImportError:
log = logging.getLogger(__name__)
DEFAULT_FTYPE = 'auto'
DEFAULT_DBKEY = 'hg17'
DEFAULT_INTERACTOR = "api" # Default mechanism test code uses for interacting with Galaxy instance.
DEFAULT_MAX_SECS = None
@nottest
def parse_tests(tool, tests_source):
@@ -31,12 +27,12 @@ def parse_tests(tool, tests_source):
raw_tests_dict = tests_source.parse_tests_to_dict()
tests = []
for i, raw_test_dict in enumerate(raw_tests_dict.get('tests', [])):
test = builder_from_tool_object(tool, i, raw_test_dict)
test = description_from_tool_object(tool, i, raw_test_dict)
tests.append(test)
return tests
def builder_from_tool_object(tool, test_index, raw_test_dict):
def description_from_tool_object(tool, test_index, raw_test_dict):
required_files = []
num_outputs = raw_test_dict.get('expect_num_outputs', None)
@@ -69,64 +65,7 @@ def builder_from_tool_object(tool, test_index, raw_test_dict):
"exception": str(e),
}
return ToolTestBuilder(processed_test_dict)
class ToolTestBuilder(object):
"""
Encapsulates information about a tool test, and allows creation of a
dynamic TestCase class (the unittest framework is very class oriented,
doing dynamic tests in this way allows better integration)
"""
def __init__(self, processed_test_dict):
test_index = processed_test_dict["test_index"]
name = processed_test_dict.get('name', 'Test-%d' % (test_index + 1))
maxseconds = processed_test_dict.get('maxseconds', DEFAULT_MAX_SECS)
if maxseconds is not None:
maxseconds = int(maxseconds)
self.tool_id = processed_test_dict["tool_id"]
self.name = name
self.maxseconds = maxseconds
self.required_files = processed_test_dict.get("required_files", [])
self.inputs = processed_test_dict.get("inputs", {})
self.outputs = processed_test_dict.get("outputs", [])
self.num_outputs = processed_test_dict.get("num_outputs", 0)
self.error = processed_test_dict.get("error", False)
self.exception = processed_test_dict.get("exception", None)
self.output_collections = processed_test_dict.get("output_collections", None)
self.command_line = processed_test_dict.get("command", None)
self.stdout = processed_test_dict.get("stdout", None)
self.stderr = processed_test_dict.get("stderr", None)
self.expect_exit_code = processed_test_dict.get("expect_exit_code", None)
self.expect_failure = processed_test_dict.get("expect_failure", False)
self.md5 = processed_test_dict.get("md5", None)
def test_data(self):
"""
Iterator over metadata representing the required files for upload.
"""
return test_data_iter(self.required_files)
def to_dict(self):
return {
"inputs": self.inputs,
"outputs": self.outputs,
"output_collections": self.output_collections,
"num_outputs": self.num_outputs,
"command_line": self.command_line,
"stdout": self.stdout,
"stderr": self.stderr,
"expect_exit_code": self.expect_exit_code,
"expect_failure": self.expect_failure,
"md5": self.md5,
"name": self.name,
"test_index": self.test_index,
"tool_id": self.tool_id,
}
return ToolTestDescription(processed_test_dict)
def _process_raw_inputs(tool, tool_inputs, raw_inputs, required_files, parent_context=None):
@@ -316,30 +255,6 @@ def _process_bool_param_value(param, param_value):
return [processed_value] if was_list else processed_value
@nottest
def test_data_iter(required_files):
for fname, extra in required_files:
data_dict = dict(
fname=fname,
metadata=extra.get('metadata', []),
composite_data=extra.get('composite_data', []),
ftype=extra.get('ftype', DEFAULT_FTYPE),
dbkey=extra.get('dbkey', DEFAULT_DBKEY),
)
edit_attributes = extra.get('edit_attributes', [])
# currently only renaming is supported
for edit_att in edit_attributes:
if edit_att.get('type', None) == 'name':
new_name = edit_att.get('value', None)
assert new_name, 'You must supply the new dataset name as the value tag of the edit_attributes tag'
data_dict['name'] = new_name
else:
raise Exception('edit_attributes type (%s) is unimplemented' % edit_att.get('type', None))
yield data_dict
def require_file(name, value, extra, required_files):
if (value, extra) not in required_files:
required_files.append((value, extra)) # these files will be uploaded
lib/galaxy/tools/verify/interactor.py
+127 -5
View File
@@ -8,6 +8,11 @@ import time
from json import dumps
from logging import getLogger
try:
from nose.tools import nottest
except ImportError:
def nottest(x):
return x
try:
import requests
except ImportError:
@@ -15,7 +20,7 @@ except ImportError:
from six import StringIO, text_type
from galaxy import util
from galaxy.tools.parser.interface import TestCollectionDef
from galaxy.tools.parser.interface import TestCollectionDef, TestCollectionOutputDef
from galaxy.util.bunch import Bunch
from galaxy.util.odict import odict
from .asserts import verify_assertions
@@ -31,6 +36,10 @@ UPLOAD_ASYNC = util.asbool(os.environ.get("GALAXY_TEST_UPLOAD_ASYNC", True))
ERROR_MESSAGE_DATASET_SEP = "--------------------------------------"
DEFAULT_TOOL_TEST_WAIT = os.environ.get("GALAXY_TEST_DEFAULT_WAIT", 86400)
DEFAULT_FTYPE = 'auto'
DEFAULT_DBKEY = 'hg17'
DEFAULT_MAX_SECS = None
def stage_data_in_history(galaxy_interactor, tool_id, all_test_data, history):
# Upload any needed files
@@ -67,6 +76,16 @@ class GalaxyInteractorApi(object):
test_user = self.ensure_user_with_email(test_user)
return self._post("users/%s/api_key" % test_user['id'], key=admin_key).json()
def get_tools(self):
response = self._get("tools?in_panel=false")
assert response.status_code == 200, "Non 200 response from tool index API. [%s]" % response.content
return response.json()
def get_tool_tests(self, tool_id):
response = self._get("tools/%s/test_data" % tool_id)
assert response.status_code == 200, "Non 200 response from tool test API. [%s]" % response.content
return response.json()
def verify_output(self, history_id, jobs, output_data, output_testdef, tool_id, maxseconds):
outfile = output_testdef.outfile
attributes = output_testdef.attributes
@@ -202,6 +221,7 @@ class GalaxyInteractorApi(object):
history_json = self._post("histories", {"name": "test_history"}).json()
return history_json['id']
@nottest
def test_data_path(self, tool_id, filename):
return self._get("tools/%s/test_data_path?filename=%s" % (tool_id, filename)).json()
@@ -220,14 +240,17 @@ class GalaxyInteractorApi(object):
"file_type": test_data['ftype'],
"dbkey": test_data['dbkey'],
}
for elem in test_data.get('metadata', []):
tool_input["files_metadata|%s" % elem.get('name')] = elem.get('value')
metadata = test_data.get("metadata", {})
if not hasattr(metadata, "items"):
raise Exception("Invalid metadata description found for input [%s] - [%s]" % (fname, metadata))
for name, value in test_data.get('metadata', {}).items():
tool_input["files_metadata|%s" % name] = value
composite_data = test_data['composite_data']
if composite_data:
files = {}
for i, composite_file in enumerate(composite_data):
file_name = self.test_data_path(tool_id, composite_file.get("value"))
for i, file_name in enumerate(composite_data):
file_name = self.test_data_path(tool_id, file_name)
files["files_%s|file_data" % i] = open(file_name, 'rb')
tool_input.update({
"files_%d|type" % i: "upload_dataset",
@@ -830,3 +853,102 @@ class JobOutputsError(AssertionError):
super(JobOutputsError, self).__init__(big_message)
self.job_stdio = job_stdio
self.output_exceptions = output_exceptions
class ToolTestDescription(object):
"""
Encapsulates information about a tool test, and allows creation of a
dynamic TestCase class (the unittest framework is very class oriented,
doing dynamic tests in this way allows better integration)
"""
def __init__(self, processed_test_dict):
test_index = processed_test_dict["test_index"]
name = processed_test_dict.get('name', 'Test-%d' % (test_index + 1))
maxseconds = processed_test_dict.get('maxseconds', DEFAULT_MAX_SECS)
if maxseconds is not None:
maxseconds = int(maxseconds)
self.test_index = test_index
self.tool_id = processed_test_dict["tool_id"]
self.name = name
self.maxseconds = maxseconds
self.required_files = processed_test_dict.get("required_files", [])
inputs = processed_test_dict.get("inputs", {})
loaded_inputs = {}
for key, value in inputs.items():
if isinstance(value, dict) and value.get("model_class"):
loaded_inputs[key] = TestCollectionDef.from_dict(value)
else:
loaded_inputs[key] = value
self.inputs = loaded_inputs
self.outputs = processed_test_dict.get("outputs", [])
self.num_outputs = processed_test_dict.get("num_outputs", 0)
self.error = processed_test_dict.get("error", False)
self.exception = processed_test_dict.get("exception", None)
self.output_collections = map(TestCollectionOutputDef.from_dict, processed_test_dict.get("output_collections", []))
self.command_line = processed_test_dict.get("command", None)
self.stdout = processed_test_dict.get("stdout", None)
self.stderr = processed_test_dict.get("stderr", None)
self.expect_exit_code = processed_test_dict.get("expect_exit_code", None)
self.expect_failure = processed_test_dict.get("expect_failure", False)
self.md5 = processed_test_dict.get("md5", None)
def test_data(self):
"""
Iterator over metadata representing the required files for upload.
"""
return test_data_iter(self.required_files)
def to_dict(self):
inputs_dict = {}
for key, value in self.inputs.items():
if hasattr(value, "to_dict"):
inputs_dict[key] = value.to_dict()
else:
inputs_dict[key] = value
return {
"inputs": inputs_dict,
"outputs": self.outputs,
"output_collections": map(lambda o: o.to_dict(), self.output_collections),
"num_outputs": self.num_outputs,
"command_line": self.command_line,
"stdout": self.stdout,
"stderr": self.stderr,
"expect_exit_code": self.expect_exit_code,
"expect_failure": self.expect_failure,
"md5": self.md5,
"name": self.name,
"test_index": self.test_index,
"tool_id": self.tool_id,
"required_files": self.required_files,
}
@nottest
def test_data_iter(required_files):
for fname, extra in required_files:
data_dict = dict(
fname=fname,
metadata=extra.get('metadata', {}),
composite_data=extra.get('composite_data', []),
ftype=extra.get('ftype', DEFAULT_FTYPE),
dbkey=extra.get('dbkey', DEFAULT_DBKEY),
)
edit_attributes = extra.get('edit_attributes', [])
# currently only renaming is supported
for edit_att in edit_attributes:
if edit_att.get('type', None) == 'name':
new_name = edit_att.get('value', None)
assert new_name, 'You must supply the new dataset name as the value tag of the edit_attributes tag'
data_dict['name'] = new_name
else:
raise Exception('edit_attributes type (%s) is unimplemented' % edit_att.get('type', None))
yield data_dict
lib/galaxy/webapps/galaxy/api/tools.py
+24
View File
@@ -6,10 +6,13 @@ from six.moves.urllib.parse import unquote_plus
import galaxy.queue_worker
from galaxy import exceptions, managers, util, web
from galaxy.managers.collections_util import dictify_dataset_collection_instance
from galaxy.util.json import safe_dumps
from galaxy.util.odict import odict
from galaxy.visualization.genomes import GenomeRegion
from galaxy.web import _future_expose_api as expose_api
from galaxy.web import _future_expose_api_anonymous as expose_api_anonymous
from galaxy.web import _future_expose_api_anonymous_and_sessionless as expose_api_anonymous_and_sessionless
from galaxy.web import _future_expose_api_raw_anonymous_and_sessionless as expose_api_raw_anonymous_and_sessionless
from galaxy.web.base.controller import BaseAPIController
from galaxy.web.base.controller import UsesVisualizationMixin
@@ -111,6 +114,27 @@ class ToolsController(BaseAPIController, UsesVisualizationMixin):
tool = self._get_tool(id, tool_version=tool_version, user=trans.user)
return tool.test_data_path(kwd.get("filename"))
@expose_api_raw_anonymous_and_sessionless
def test_data(self, trans, id, **kwd):
"""
GET /api/tools/{tool_id}/test_data
"""
# TODO: eliminate copy and paste with above code.
if 'payload' in kwd:
kwd = kwd.get('payload')
tool_version = kwd.get('tool_version', None)
tool = self._get_tool(id, tool_version=tool_version, user=trans.user)
# Encode in this method to handle odict objects in tool representation.
def json_encodeify(obj):
if isinstance(obj, odict):
return dict(obj)
else:
return obj
result = [t.to_dict() for t in tool.tests]
return safe_dumps(result, default=json_encodeify)
@expose_api
@web.require_admin
def reload(self, trans, id, **kwd):
lib/galaxy/webapps/galaxy/buildapp.py
+1
View File
@@ -275,6 +275,7 @@ def populate_api_routes(webapp, app):
webapp.mapper.connect('/api/tools/{id:.+?}/build', action='build', controller="tools")
webapp.mapper.connect('/api/tools/{id:.+?}/reload', action='reload', controller="tools")
webapp.mapper.connect('/api/tools/{id:.+?}/test_data_path', action='test_data_path', controller="tools")
webapp.mapper.connect('/api/tools/{id:.+?}/test_data', action='test_data', controller="tools")
webapp.mapper.connect('/api/tools/{id:.+?}/diagnostics', action='diagnostics', controller="tools")
webapp.mapper.connect('/api/tools/{id:.+?}/citations', action='citations', controller="tools")
webapp.mapper.connect('/api/tools/{id:.+?}/download', action='download', controller="tools")
test/api/test_tools.py
+38
View File
@@ -135,6 +135,44 @@ class ToolsTestCase(api.ApiTestCase):
self._assert_has_keys(tool_info, "inputs", "outputs", "panel_section_id")
return tool_info
@skip_without_tool("composite_output")
def test_test_data_composite_output(self):
test_data_response = self._get("tools/%s/test_data" % "composite_output")
assert test_data_response.status_code == 200
test_data = test_data_response.json()
assert len(test_data) == 1
test_case = test_data[0]
self._assert_has_keys(test_case, "inputs", "outputs", "output_collections", "required_files")
assert len(test_case["inputs"]) == 1, test_case
# input0 = next(iter(test_case["inputs"].values()))
@skip_without_tool("collection_two_paired")
def test_test_data_collection_two_paired(self):
test_data_response = self._get("tools/%s/test_data" % "collection_two_paired")
assert test_data_response.status_code == 200
test_data = test_data_response.json()
assert len(test_data) == 2
test_case = test_data[0]
self._assert_has_keys(test_case, "inputs", "outputs", "output_collections", "required_files")
assert len(test_case["inputs"]) == 3, test_case
@skip_without_tool("collection_nested_test")
def test_test_data_collection_nested(self):
test_data_response = self._get("tools/%s/test_data" % "collection_nested_test")
assert test_data_response.status_code == 200
test_data = test_data_response.json()
assert len(test_data) == 2
test_case = test_data[0]
self._assert_has_keys(test_case, "inputs", "outputs", "output_collections", "required_files")
assert len(test_case["inputs"]) == 1, test_case
@skip_without_tool("simple_constructs_y")
def test_test_data_yaml_tools(self):
test_data_response = self._get("tools/%s/test_data" % "simple_constructs_y")
assert test_data_response.status_code == 200
test_data = test_data_response.json()
assert len(test_data) == 3
def test_unzip_collection(self):
with self.dataset_populator.test_history() as history_id:
hdca_id = self.__build_pair(history_id, ["123", "456"])
test/base/driver_util.py
+32 -11
View File
@@ -28,6 +28,7 @@ from six.moves.urllib.parse import urlparse
from galaxy.app import UniverseApplication as GalaxyUniverseApplication
from galaxy.config import LOGGING_CONFIG_DEFAULT
from galaxy.tools.verify.interactor import GalaxyInteractorApi, ToolTestDescription, verify_tool
from galaxy.util import asbool, download_to_file
from galaxy.util.properties import load_app_properties
from galaxy.web import buildapp
@@ -908,17 +909,17 @@ class GalaxyTestDriver(TestDriver):
return functional.test_toolbox
def run_tool_test(self, tool_id, index=0, resource_parameters={}):
import functional.test_toolbox
functional.test_toolbox.toolbox = self.app.toolbox
tool = self.app.toolbox.get_tool(tool_id)
testdef = tool.tests[index]
test_case_cls = functional.test_toolbox.ToolTestCase
test_case = test_case_cls(methodName="setUp") # NO-OP
test_case.tool_id = tool_id
test_case.master_api_key = get_master_api_key()
test_case.user_api_key = get_user_api_key()
test_case.setUp()
test_case.do_it(testdef, resource_parameters=resource_parameters)
host, port, url = target_url_parts()
galaxy_interactor_kwds = {
"galaxy_url": url,
"master_api_key": get_master_api_key(),
"api_key": get_user_api_key(),
"keep_outputs_dir": None,
}
galaxy_interactor = GalaxyInteractorApi(**galaxy_interactor_kwds)
tool_test_dicts = galaxy_interactor.get_tool_tests(tool_id)
testdef = ToolTestDescription(tool_test_dicts[index])
verify_tool(testdef, tool_id, galaxy_interactor, resource_parameters=resource_parameters)
def drive_test(test_driver_class):
@@ -927,6 +928,24 @@ def drive_test(test_driver_class):
sys.exit(test_driver.run())
def setup_keep_outdir():
keep_outdir = os.environ.get('GALAXY_TEST_SAVE', '')
if keep_outdir > '':
try:
os.makedirs(keep_outdir)
except Exception:
pass
return keep_outdir
def target_url_parts():
host = os.environ.get('GALAXY_TEST_HOST')
port = os.environ.get('GALAXY_TEST_PORT')
default_url = "http://%s:%s" % (host, port)
url = os.environ.get('GALAXY_TEST_EXTERNAL', default_url)
return host, port, url
__all__ = (
"copy_database_template",
"build_logger",
@@ -937,7 +956,9 @@ __all__ = (
"database_conf",
"get_webapp_global_conf",
"nose_config_and_run",
"setup_keep_outdir",
"setup_galaxy_config",
"target_url_parts",
"TestDriver",
"wait_for_http_server",
)
test/base/testcase.py
+3 -11
View File
@@ -6,7 +6,7 @@ import unittest
from galaxy.tools.verify.test_data import TestDataResolver
from galaxy.web import security
from .driver_util import GalaxyTestDriver
from .driver_util import GalaxyTestDriver, setup_keep_outdir, target_url_parts
log = logging.getLogger(__name__)
@@ -17,17 +17,9 @@ class FunctionalTestCase(unittest.TestCase):
# 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.host, self.port, self.url = target_url_parts()
self.test_data_resolver = TestDataResolver()
self.keepOutdir = os.environ.get('GALAXY_TEST_SAVE', '')
if self.keepOutdir > '':
try:
os.makedirs(self.keepOutdir)
except Exception:
pass
self.keepOutdir = setup_keep_outdir()
@classmethod
def setUpClass(cls):
test/functional/test_toolbox.py
+34 -14
View File
@@ -9,9 +9,9 @@ except ImportError:
def nottest(x):
return x
from base.interactor import TestCaseGalaxyInteractor
from base.driver_util import setup_keep_outdir, target_url_parts
from galaxy.tools import DataManagerTool # noqa: I201
from galaxy.tools.verify.interactor import verify_tool # noqa: I201
from galaxy.tools.verify.interactor import GalaxyInteractorApi, ToolTestDescription, verify_tool # noqa: I201
from .twilltestcase import TwillTestCase
log = logging.getLogger(__name__)
@@ -37,7 +37,7 @@ class ToolTestCase(TwillTestCase):
self._handle_test_def_errors(testdef)
galaxy_interactor = TestCaseGalaxyInteractor(self)
galaxy_interactor = self.galaxy_interactor
verify_tool(testdef, tool_id, galaxy_interactor, resource_parameters=resource_parameters)
@@ -60,8 +60,17 @@ def build_tests(app=None, testing_shed_tools=False, master_api_key=None, user_ap
classes for all of its tests and put them into this modules globals() so
they can be discovered by nose.
"""
if app is None:
return
# galaxy_interactor = None
# if app is None:
host, port, url = target_url_parts()
keep_outputs_dir = setup_keep_outdir()
galaxy_interactor_kwds = {
"galaxy_url": url,
"master_api_key": master_api_key,
"api_key": user_api_key,
"keep_outputs_dir": keep_outputs_dir,
}
galaxy_interactor = GalaxyInteractorApi(**galaxy_interactor_kwds)
# Push all the toolbox tests to module level
G = globals()
@@ -70,19 +79,29 @@ def build_tests(app=None, testing_shed_tools=False, master_api_key=None, user_ap
for key, val in G.items():
if key.startswith('TestForTool_'):
del G[key]
for i, tool_id in enumerate(app.toolbox.tools_by_id):
tool = app.toolbox.get_tool(tool_id)
if isinstance(tool, TOOL_TYPES_NO_TEST):
# if app:
# tool_ids = app.toolbox.tools_by_id.keys()
# else:
tools = galaxy_interactor.get_tools()
for tool in tools:
if tool.get("form_style", None) != "regular":
# We do not test certain types of tools (e.g. Data Manager tools) as part of ToolTestCase
continue
if tool.tests:
tool_id = tool.id
tool_id = tool["id"]
if not tool_id:
continue
tool_name = tool["name"]
tool_test_dicts = galaxy_interactor.get_tool_tests(tool_id)
if tool_test_dicts:
# Create a new subclass of ToolTestCase, dynamically adding methods
# named test_tool_XXX that run each test defined in the tool config.
name = "TestForTool_" + tool.id.replace(' ', '_')
name = "TestForTool_" + tool_id.replace(' ', '_')
baseclasses = (ToolTestCase, )
namespace = dict()
for j, testdef in enumerate(tool.tests):
for j, tool_test_dict in enumerate(tool_test_dicts):
testdef = ToolTestDescription(tool_test_dict)
test_function_name = 'test_tool_%06d' % j
def make_test_method(td):
@@ -93,12 +112,13 @@ def build_tests(app=None, testing_shed_tools=False, master_api_key=None, user_ap
return test_tool
test_method = make_test_method(testdef)
test_method.__doc__ = "%s ( %s ) > %s" % (tool.name, tool.id, testdef.name)
test_method.__doc__ = "%s ( %s ) > %s" % (tool_name, tool_id, testdef.name)
namespace[test_function_name] = test_method
namespace['tool_id'] = tool_id
namespace["galaxy_interactor"] = galaxy_interactor
namespace['master_api_key'] = master_api_key
namespace['user_api_key'] = user_api_key
# The new.classobj function returns a new class object, with name name, derived
# from baseclasses (which should be a tuple of classes) and with namespace dict.
new_class_obj = new.classobj(name, baseclasses, namespace)
new_class_obj = new.classobj(str(name), baseclasses, namespace)
G[name] = new_class_obj