Introduce a TestDriver class to reduce duplication between test drivers.

Classes allow de-duplicating logic and state organization in a different way than methods. Use a common setup, run, tear down paradigm for these tests.
This commit is contained in:
John Chilton
2016-03-29 08:04:28 -04:00
parent 8538c0ed77
commit 6faef4df2e
3 changed files with 207 additions and 188 deletions
+42 -58
View File
@@ -14,63 +14,61 @@ galaxy_root = os.path.abspath(os.path.join(os.path.dirname(__file__), os.path.pa
sys.path[1:1] = [ os.path.join( galaxy_root, "lib" ), os.path.join( galaxy_root, "test" ) ]
from base import driver_util
driver_util.configure_environment()
log = driver_util.build_logger()
from base.api_util import get_master_api_key, get_user_api_key
from galaxy.web import buildapp
def main():
"""Entry point for test driver script."""
# ---- Configuration ------------------------------------------------------
testing_migrated_tools = _check_arg('-migrated')
testing_installed_tools = _check_arg('-installed')
testing_framework_tools = _check_arg('-framework')
testing_data_manager = _check_arg('-data_managers')
testing_workflow = _check_arg('-workflow')
testing_shed_tools = testing_migrated_tools or testing_installed_tools
class GalaxyTestDriver(driver_util.TestDriver):
"""Instantial a Galaxy-style nose TestDriver for testing Galaxy."""
datatypes_conf_override = None
default_tool_conf = None
def setup(self):
"""Setup a Galaxy server for functional test (if needed)."""
# ---- Configuration ------------------------------------------------------
testing_migrated_tools = _check_arg('-migrated')
testing_installed_tools = _check_arg('-installed')
testing_framework_tools = _check_arg('-framework')
testing_data_manager = _check_arg('-data_managers')
testing_workflow = _check_arg('-workflow')
testing_shed_tools = testing_migrated_tools or testing_installed_tools
if testing_framework_tools:
default_tool_conf = driver_util.FRAMEWORK_SAMPLE_TOOLS_CONF
datatypes_conf_override = driver_util.FRAMEWORK_DATATYPES_CONF
datatypes_conf_override = None
default_tool_conf = None
external_galaxy = os.environ.get('GALAXY_TEST_EXTERNAL', None)
if testing_framework_tools:
default_tool_conf = driver_util.FRAMEWORK_SAMPLE_TOOLS_CONF
datatypes_conf_override = driver_util.FRAMEWORK_DATATYPES_CONF
galaxy_test_tmp_dir = driver_util.get_galaxy_test_tmp_dir()
external_galaxy = os.environ.get('GALAXY_TEST_EXTERNAL', None)
app = None
server_wrapper = None
galaxy_test_tmp_dir = driver_util.get_galaxy_test_tmp_dir()
self.temp_directories.append(galaxy_test_tmp_dir)
if external_galaxy is None:
tempdir = tempfile.mkdtemp( dir=galaxy_test_tmp_dir )
# Configure the database path.
galaxy_db_path = driver_util.database_files_path(tempdir)
galaxy_config = driver_util.setup_galaxy_config(
galaxy_db_path,
use_test_file_dir=not testing_shed_tools,
default_install_db_merged=True,
default_tool_conf=default_tool_conf,
datatypes_conf=datatypes_conf_override,
)
if external_galaxy is None:
tempdir = tempfile.mkdtemp( dir=galaxy_test_tmp_dir )
# Configure the database path.
galaxy_db_path = driver_util.database_files_path(tempdir)
galaxy_config = driver_util.setup_galaxy_config(
galaxy_db_path,
use_test_file_dir=not testing_shed_tools,
default_install_db_merged=True,
default_tool_conf=default_tool_conf,
datatypes_conf=datatypes_conf_override,
)
# ---- Build Application --------------------------------------------------
app = driver_util.build_galaxy_app(galaxy_config)
server_wrapper = driver_util.launch_server(
app,
buildapp.app_factory,
galaxy_config,
)
log.info("Functional tests will be run against %s:%s" % (server_wrapper.host, server_wrapper.port))
else:
log.info("Functional tests will be run against %s" % external_galaxy)
# ---- Build Application --------------------------------------------------
app = driver_util.build_galaxy_app(galaxy_config)
server_wrapper = driver_util.launch_server(
app,
buildapp.app_factory,
galaxy_config,
)
self.server_wrappers.append(server_wrapper)
log.info("Functional tests will be run against %s:%s" % (server_wrapper.host, server_wrapper.port))
else:
log.info("Functional tests will be run against %s" % external_galaxy)
# ---- Find tests ---------------------------------------------------------
success = False
try:
if testing_shed_tools:
driver_util.setup_shed_tools_for_test(
app,
@@ -106,20 +104,6 @@ def main():
master_api_key=get_master_api_key(),
user_api_key=get_user_api_key(),
)
success = driver_util.nose_config_and_run()
except:
log.exception( "Failure running tests" )
log.info( "Shutting down" )
# ---- Tear down -----------------------------------------------------------
if server_wrapper is not None:
server_wrapper.stop()
server_wrapper = None
driver_util.cleanup_directory(galaxy_test_tmp_dir)
if success:
return 0
else:
return 1
def _check_arg( name ):
@@ -132,4 +116,4 @@ def _check_arg( name ):
return ret_val
if __name__ == "__main__":
sys.exit( main() )
driver_util.drive_test(GalaxyTestDriver)
+53 -2
View File
@@ -529,11 +529,61 @@ def launch_server(app, webapp_factory, kwargs, prefix="GALAXY"):
)
class TestDriver(object):
"""Responsible for the life-cycle of a Galaxy-style functional test.
Sets up servers, configures tests, runs nose, and tears things
down. This is somewhat like a Python TestCase - but different
because it is meant to provide a main() endpoint.
"""
def __init__(self):
"""Setup tracked resources."""
self.server_wrappers = []
self.temp_directories = []
def setup(self):
"""Called before tests are built."""
def build_tests(self):
"""After environment is setup, setup nose tests."""
def tear_down(self):
"""Cleanup resources tracked by this object."""
for server_wrapper in self.server_wrappers:
server_wrapper.stop()
for temp_directory in self.temp_directories:
cleanup_directory(temp_directory)
def run(self):
"""Driver whole test.
Setup environment, build tests (if needed), run test,
and finally cleanup resources.
"""
configure_environment()
self.setup()
self.build_tests()
try:
success = nose_config_and_run()
return 0 if success else 1
except Exception as e:
log.info("Failure running tests")
raise e
finally:
log.info( "Shutting down")
self.tear_down()
def drive_test(test_driver_class):
"""Instantiate driver class, run, and exit appropriately."""
sys.exit(test_driver_class().run())
__all__ = [
"cleanup_directory",
"configure_environment",
"copy_database_template",
"build_logger",
"drive_test",
"FRAMEWORK_UPLOAD_TOOL_CONF",
"FRAMEWORK_SAMPLE_TOOLS_CONF",
"FRAMEWORK_DATATYPES_CONF",
@@ -542,5 +592,6 @@ __all__ = [
"nose_config_and_run",
"setup_galaxy_config",
"setup_shed_tools_for_test",
"TestDriver",
"wait_for_http_server",
]
+112 -128
View File
@@ -15,9 +15,7 @@ galaxy_root = os.path.abspath(os.path.join(os.path.dirname(__file__), os.path.pa
sys.path[0:1] = [ os.path.join( galaxy_root, "lib" ), os.path.join( galaxy_root, "test" ) ]
from base import driver_util
driver_util.configure_environment()
log = driver_util.build_logger()
tool_shed_test_tmp_dir = driver_util.setup_tool_shed_tmp_dir()
# This is for the tool shed application.
from galaxy.webapps.tool_shed import buildapp as toolshedbuildapp
@@ -48,138 +46,124 @@ shed_data_manager_conf_xml_template = '''<?xml version="1.0"?>
'''
def main():
"""Entry point for test driver script."""
# ---- Configuration ------------------------------------------------------
if not os.path.isdir( tool_shed_test_tmp_dir ):
os.mkdir( tool_shed_test_tmp_dir )
shed_db_path = driver_util.database_files_path(tool_shed_test_tmp_dir, prefix="TOOL_SHED")
shed_tool_data_table_conf_file = os.environ.get( 'TOOL_SHED_TEST_TOOL_DATA_TABLE_CONF', os.path.join( tool_shed_test_tmp_dir, 'shed_tool_data_table_conf.xml' ) )
galaxy_shed_data_manager_conf_file = os.environ.get( 'GALAXY_SHED_DATA_MANAGER_CONF', os.path.join( tool_shed_test_tmp_dir, 'test_shed_data_manager_conf.xml' ) )
default_tool_data_table_config_path = os.path.join( tool_shed_test_tmp_dir, 'tool_data_table_conf.xml' )
galaxy_shed_tool_conf_file = os.environ.get( 'GALAXY_TEST_SHED_TOOL_CONF', os.path.join( tool_shed_test_tmp_dir, 'test_shed_tool_conf.xml' ) )
galaxy_migrated_tool_conf_file = os.environ.get( 'GALAXY_TEST_MIGRATED_TOOL_CONF', os.path.join( tool_shed_test_tmp_dir, 'test_migrated_tool_conf.xml' ) )
galaxy_tool_sheds_conf_file = os.environ.get( 'GALAXY_TEST_TOOL_SHEDS_CONF', os.path.join( tool_shed_test_tmp_dir, 'test_sheds_conf.xml' ) )
if 'GALAXY_TEST_TOOL_DATA_PATH' in os.environ:
tool_data_path = os.environ.get( 'GALAXY_TEST_TOOL_DATA_PATH' )
else:
tool_data_path = tempfile.mkdtemp( dir=tool_shed_test_tmp_dir )
os.environ[ 'GALAXY_TEST_TOOL_DATA_PATH' ] = tool_data_path
galaxy_db_path = driver_util.database_files_path(tool_shed_test_tmp_dir)
shed_file_path = os.path.join( shed_db_path, 'files' )
hgweb_config_file_path = tempfile.mkdtemp( dir=tool_shed_test_tmp_dir )
new_repos_path = tempfile.mkdtemp( dir=tool_shed_test_tmp_dir )
galaxy_shed_tool_path = tempfile.mkdtemp( dir=tool_shed_test_tmp_dir )
galaxy_migrated_tool_path = tempfile.mkdtemp( dir=tool_shed_test_tmp_dir )
hgweb_config_dir = hgweb_config_file_path
os.environ[ 'TEST_HG_WEB_CONFIG_DIR' ] = hgweb_config_dir
print "Directory location for hgweb.config:", hgweb_config_dir
toolshed_database_conf = driver_util.database_conf(shed_db_path, prefix="TOOL_SHED")
kwargs = dict( admin_users='test@bx.psu.edu',
allow_user_creation=True,
allow_user_deletion=True,
datatype_converters_config_file='datatype_converters_conf.xml.sample',
file_path=shed_file_path,
hgweb_config_dir=hgweb_config_dir,
job_queue_workers=5,
id_secret='changethisinproductiontoo',
log_destination="stdout",
new_file_path=new_repos_path,
running_functional_tests=True,
shed_tool_data_table_config=shed_tool_data_table_conf_file,
smtp_server='smtp.dummy.string.tld',
email_from='functional@localhost',
template_path='templates',
tool_parse_help=False,
use_heartbeat=False )
kwargs.update(toolshed_database_conf)
# Generate the tool_data_table_conf.xml file.
file( default_tool_data_table_config_path, 'w' ).write( tool_data_table_conf_xml_template )
# Generate the shed_tool_data_table_conf.xml file.
file( shed_tool_data_table_conf_file, 'w' ).write( tool_data_table_conf_xml_template )
os.environ[ 'TOOL_SHED_TEST_TOOL_DATA_TABLE_CONF' ] = shed_tool_data_table_conf_file
# ---- Build Tool Shed Application --------------------------------------------------
toolshedapp = driver_util.build_shed_app(kwargs)
class ToolShedTestDriver(driver_util.TestDriver):
"""Instantial a Galaxy-style nose TestDriver for testing the tool shed."""
# ---- Run tool shed webserver ------------------------------------------------------
# TODO: Needed for hg middleware ('lib/galaxy/webapps/tool_shed/framework/middleware/hg.py')
kwargs['global_conf']['database_connection'] = kwargs["database_connection"]
tool_shed_server_wrapper = driver_util.launch_server(
toolshedapp,
toolshedbuildapp.app_factory,
kwargs,
prefix="TOOL_SHED",
)
tool_shed_test_host = tool_shed_server_wrapper.host
tool_shed_test_port = tool_shed_server_wrapper.port
log.info( "Functional tests will be run against %s:%s" % ( tool_shed_test_host, tool_shed_test_port ) )
def setup(self):
"""Entry point for test driver script."""
# ---- Configuration ------------------------------------------------------
tool_shed_test_tmp_dir = driver_util.setup_tool_shed_tmp_dir()
if not os.path.isdir( tool_shed_test_tmp_dir ):
os.mkdir( tool_shed_test_tmp_dir )
self.temp_directories.append(tool_shed_test_tmp_dir)
shed_db_path = driver_util.database_files_path(tool_shed_test_tmp_dir, prefix="TOOL_SHED")
shed_tool_data_table_conf_file = os.environ.get( 'TOOL_SHED_TEST_TOOL_DATA_TABLE_CONF', os.path.join( tool_shed_test_tmp_dir, 'shed_tool_data_table_conf.xml' ) )
galaxy_shed_data_manager_conf_file = os.environ.get( 'GALAXY_SHED_DATA_MANAGER_CONF', os.path.join( tool_shed_test_tmp_dir, 'test_shed_data_manager_conf.xml' ) )
default_tool_data_table_config_path = os.path.join( tool_shed_test_tmp_dir, 'tool_data_table_conf.xml' )
galaxy_shed_tool_conf_file = os.environ.get( 'GALAXY_TEST_SHED_TOOL_CONF', os.path.join( tool_shed_test_tmp_dir, 'test_shed_tool_conf.xml' ) )
galaxy_migrated_tool_conf_file = os.environ.get( 'GALAXY_TEST_MIGRATED_TOOL_CONF', os.path.join( tool_shed_test_tmp_dir, 'test_migrated_tool_conf.xml' ) )
galaxy_tool_sheds_conf_file = os.environ.get( 'GALAXY_TEST_TOOL_SHEDS_CONF', os.path.join( tool_shed_test_tmp_dir, 'test_sheds_conf.xml' ) )
if 'GALAXY_TEST_TOOL_DATA_PATH' in os.environ:
tool_data_path = os.environ.get( 'GALAXY_TEST_TOOL_DATA_PATH' )
else:
tool_data_path = tempfile.mkdtemp( dir=tool_shed_test_tmp_dir )
os.environ[ 'GALAXY_TEST_TOOL_DATA_PATH' ] = tool_data_path
galaxy_db_path = driver_util.database_files_path(tool_shed_test_tmp_dir)
shed_file_path = os.path.join( shed_db_path, 'files' )
hgweb_config_file_path = tempfile.mkdtemp( dir=tool_shed_test_tmp_dir )
new_repos_path = tempfile.mkdtemp( dir=tool_shed_test_tmp_dir )
galaxy_shed_tool_path = tempfile.mkdtemp( dir=tool_shed_test_tmp_dir )
galaxy_migrated_tool_path = tempfile.mkdtemp( dir=tool_shed_test_tmp_dir )
hgweb_config_dir = hgweb_config_file_path
os.environ[ 'TEST_HG_WEB_CONFIG_DIR' ] = hgweb_config_dir
print "Directory location for hgweb.config:", hgweb_config_dir
toolshed_database_conf = driver_util.database_conf(shed_db_path, prefix="TOOL_SHED")
kwargs = dict( admin_users='test@bx.psu.edu',
allow_user_creation=True,
allow_user_deletion=True,
datatype_converters_config_file='datatype_converters_conf.xml.sample',
file_path=shed_file_path,
hgweb_config_dir=hgweb_config_dir,
job_queue_workers=5,
id_secret='changethisinproductiontoo',
log_destination="stdout",
new_file_path=new_repos_path,
running_functional_tests=True,
shed_tool_data_table_config=shed_tool_data_table_conf_file,
smtp_server='smtp.dummy.string.tld',
email_from='functional@localhost',
template_path='templates',
tool_parse_help=False,
use_heartbeat=False )
kwargs.update(toolshed_database_conf)
# Generate the tool_data_table_conf.xml file.
file( default_tool_data_table_config_path, 'w' ).write( tool_data_table_conf_xml_template )
# Generate the shed_tool_data_table_conf.xml file.
file( shed_tool_data_table_conf_file, 'w' ).write( tool_data_table_conf_xml_template )
os.environ[ 'TOOL_SHED_TEST_TOOL_DATA_TABLE_CONF' ] = shed_tool_data_table_conf_file
# ---- Build Tool Shed Application --------------------------------------------------
toolshedapp = driver_util.build_shed_app(kwargs)
# ---- Optionally start up a Galaxy instance ------------------------------------------------------
if 'TOOL_SHED_TEST_OMIT_GALAXY' not in os.environ:
# Generate the shed_tool_conf.xml file.
tool_sheds_conf_template_parser = string.Template( tool_sheds_conf_xml_template )
tool_sheds_conf_xml = tool_sheds_conf_template_parser.safe_substitute( shed_url=tool_shed_test_host, shed_port=tool_shed_test_port )
file( galaxy_tool_sheds_conf_file, 'w' ).write( tool_sheds_conf_xml )
# Generate the tool_sheds_conf.xml file.
shed_tool_conf_template_parser = string.Template( shed_tool_conf_xml_template )
shed_tool_conf_xml = shed_tool_conf_template_parser.safe_substitute( shed_tool_path=galaxy_shed_tool_path )
file( galaxy_shed_tool_conf_file, 'w' ).write( shed_tool_conf_xml )
# Generate the migrated_tool_conf.xml file.
migrated_tool_conf_xml = shed_tool_conf_template_parser.safe_substitute( shed_tool_path=galaxy_migrated_tool_path )
file( galaxy_migrated_tool_conf_file, 'w' ).write( migrated_tool_conf_xml )
os.environ[ 'GALAXY_TEST_SHED_TOOL_CONF' ] = galaxy_shed_tool_conf_file
# Generate shed_data_manager_conf.xml
if not os.environ.get( 'GALAXY_SHED_DATA_MANAGER_CONF' ):
open( galaxy_shed_data_manager_conf_file, 'wb' ).write( shed_data_manager_conf_xml_template )
kwargs = dict( migrated_tools_config=galaxy_migrated_tool_conf_file,
shed_data_manager_config_file=galaxy_shed_data_manager_conf_file,
shed_tool_path=galaxy_shed_tool_path,
tool_data_path=tool_data_path,
tool_sheds_config_file=galaxy_tool_sheds_conf_file )
kwargs.update(
driver_util.setup_galaxy_config(
galaxy_db_path,
use_test_file_dir=False,
default_install_db_merged=False,
default_tool_data_table_config_path=default_tool_data_table_config_path,
default_shed_tool_data_table_config=shed_tool_data_table_conf_file,
enable_tool_shed_check=True,
shed_tool_conf=galaxy_shed_tool_conf_file,
update_integrated_tool_panel=True,
)
)
print "Galaxy database connection:", kwargs["database_connection"]
# ---- Run galaxy webserver ------------------------------------------------------
galaxyapp = driver_util.build_galaxy_app(kwargs)
galaxy_server_wrapper = driver_util.launch_server(
galaxyapp,
galaxybuildapp.app_factory,
# ---- Run tool shed webserver ------------------------------------------------------
# TODO: Needed for hg middleware ('lib/galaxy/webapps/tool_shed/framework/middleware/hg.py')
kwargs['global_conf']['database_connection'] = kwargs["database_connection"]
tool_shed_server_wrapper = driver_util.launch_server(
toolshedapp,
toolshedbuildapp.app_factory,
kwargs,
prefix="TOOL_SHED",
)
log.info("Galaxy tests will be run against %s:%s" % (galaxy_server_wrapper.host, galaxy_server_wrapper.port))
self.server_wrappers.append(tool_shed_server_wrapper)
tool_shed_test_host = tool_shed_server_wrapper.host
tool_shed_test_port = tool_shed_server_wrapper.port
log.info( "Functional tests will be run against %s:%s" % ( tool_shed_test_host, tool_shed_test_port ) )
# ---- Find tests ---------------------------------------------------------
success = False
try:
success = driver_util.nose_config_and_run()
except:
log.exception( "Failure running tests" )
# ---- Optionally start up a Galaxy instance ------------------------------------------------------
if 'TOOL_SHED_TEST_OMIT_GALAXY' not in os.environ:
# Generate the shed_tool_conf.xml file.
tool_sheds_conf_template_parser = string.Template( tool_sheds_conf_xml_template )
tool_sheds_conf_xml = tool_sheds_conf_template_parser.safe_substitute( shed_url=tool_shed_test_host, shed_port=tool_shed_test_port )
file( galaxy_tool_sheds_conf_file, 'w' ).write( tool_sheds_conf_xml )
# Generate the tool_sheds_conf.xml file.
shed_tool_conf_template_parser = string.Template( shed_tool_conf_xml_template )
shed_tool_conf_xml = shed_tool_conf_template_parser.safe_substitute( shed_tool_path=galaxy_shed_tool_path )
file( galaxy_shed_tool_conf_file, 'w' ).write( shed_tool_conf_xml )
# Generate the migrated_tool_conf.xml file.
migrated_tool_conf_xml = shed_tool_conf_template_parser.safe_substitute( shed_tool_path=galaxy_migrated_tool_path )
file( galaxy_migrated_tool_conf_file, 'w' ).write( migrated_tool_conf_xml )
os.environ[ 'GALAXY_TEST_SHED_TOOL_CONF' ] = galaxy_shed_tool_conf_file
# Generate shed_data_manager_conf.xml
if not os.environ.get( 'GALAXY_SHED_DATA_MANAGER_CONF' ):
open( galaxy_shed_data_manager_conf_file, 'wb' ).write( shed_data_manager_conf_xml_template )
kwargs = dict( migrated_tools_config=galaxy_migrated_tool_conf_file,
shed_data_manager_config_file=galaxy_shed_data_manager_conf_file,
shed_tool_path=galaxy_shed_tool_path,
tool_data_path=tool_data_path,
tool_sheds_config_file=galaxy_tool_sheds_conf_file )
kwargs.update(
driver_util.setup_galaxy_config(
galaxy_db_path,
use_test_file_dir=False,
default_install_db_merged=False,
default_tool_data_table_config_path=default_tool_data_table_config_path,
default_shed_tool_data_table_config=shed_tool_data_table_conf_file,
enable_tool_shed_check=True,
shed_tool_conf=galaxy_shed_tool_conf_file,
update_integrated_tool_panel=True,
)
)
print "Galaxy database connection:", kwargs["database_connection"]
log.info( "Shutting down" )
# ---- Tear down -----------------------------------------------------------
tool_shed_server_wrapper.stop()
tool_shed_server_wrapper = None
if galaxy_server_wrapper is not None:
galaxy_server_wrapper.stop()
galaxy_server_wrapper = None
driver_util.cleanup_directory(tool_shed_test_tmp_dir)
if success:
return 0
else:
return 1
# ---- Run galaxy webserver ------------------------------------------------------
galaxyapp = driver_util.build_galaxy_app(kwargs)
galaxy_server_wrapper = driver_util.launch_server(
galaxyapp,
galaxybuildapp.app_factory,
kwargs,
)
log.info("Galaxy tests will be run against %s:%s" % (galaxy_server_wrapper.host, galaxy_server_wrapper.port))
self.server_wrappers.append(galaxy_server_wrapper)
if __name__ == "__main__":
sys.exit( main() )
driver_util.drive_test(ToolShedTestDriver)