import tempfile
import os.path
from stat import S_IXUSR
from os import makedirs, stat, symlink, chmod, environ
from shutil import rmtree
from galaxy.tools.deps import DependencyManager, INDETERMINATE_DEPENDENCY
from galaxy.tools.deps.resolvers.galaxy_packages import GalaxyPackageDependency
from galaxy.tools.deps.resolvers.modules import ModuleDependencyResolver, ModuleDependency
from galaxy.util.bunch import Bunch
from contextlib import contextmanager
from subprocess import Popen, PIPE
def test_tool_dependencies():
# Setup directories
with __test_base_path() as base_path:
for name, version, sub in [ ( "dep1", "1.0", "env.sh" ), ( "dep1", "2.0", "bin" ), ( "dep2", "1.0", None ) ]:
if sub == "bin":
p = os.path.join( base_path, name, version, "bin" )
else:
p = os.path.join( base_path, name, version )
try:
makedirs( p )
except:
pass
if sub == "env.sh":
__touch( os.path.join( p, "env.sh" ) )
dm = DependencyManager( default_base_path=base_path )
dependency = dm.find_dep( "dep1", "1.0" )
assert dependency.script == os.path.join( base_path, 'dep1', '1.0', 'env.sh' )
assert dependency.path == os.path.join( base_path, 'dep1', '1.0' )
assert dependency.version == "1.0"
dependency = dm.find_dep( "dep1", "2.0" )
assert dependency.script == None
assert dependency.path == os.path.join( base_path, 'dep1', '2.0' )
assert dependency.version == "2.0"
## Test default versions
symlink( os.path.join( base_path, 'dep1', '2.0'), os.path.join( base_path, 'dep1', 'default' ) )
dependency = dm.find_dep( "dep1", None )
assert dependency.version == "2.0"
## Test default resolve will be fall back on default package dependency
## when using the default resolver.
dependency = dm.find_dep( "dep1", "2.1" )
assert dependency.version == "2.0" # 2.0 is defined as default_version
TEST_REPO_USER = "devteam"
TEST_REPO_NAME = "bwa"
TEST_REPO_CHANGESET = "12abcd41223da"
TEST_VERSION = "0.5.9"
def test_toolshed_set_enviornment_requiremetns():
with __test_base_path() as base_path:
test_repo = __build_test_repo('set_environment')
dm = DependencyManager( default_base_path=base_path )
env_settings_dir = os.path.join(base_path, "environment_settings", TEST_REPO_NAME, TEST_REPO_USER, TEST_REPO_NAME, TEST_REPO_CHANGESET)
os.makedirs(env_settings_dir)
dependency = dm.find_dep( TEST_REPO_NAME, version=None, type='set_environment', installed_tool_dependencies=[test_repo] )
assert dependency.version == None
assert dependency.script == os.path.join(env_settings_dir, "env.sh")
def test_toolshed_package_requirements():
with __test_base_path() as base_path:
test_repo = __build_test_repo('package', version=TEST_VERSION)
dm = DependencyManager( default_base_path=base_path )
package_dir = __build_ts_test_package(base_path)
dependency = dm.find_dep( TEST_REPO_NAME, version=TEST_VERSION, type='package', installed_tool_dependencies=[test_repo] )
assert dependency.version == TEST_VERSION
assert dependency.script == os.path.join(package_dir, "env.sh")
def test_toolshed_tools_fallback_on_manual_dependencies():
with __test_base_path() as base_path:
dm = DependencyManager( default_base_path=base_path )
test_repo = __build_test_repo('package', version=TEST_VERSION)
env_path = __setup_galaxy_package_dep(base_path, "dep1", "1.0")
dependency = dm.find_dep( "dep1", version="1.0", type='package', installed_tool_dependencies=[test_repo] )
assert dependency.version == "1.0"
assert dependency.script == env_path
def test_toolshed_greater_precendence():
with __test_base_path() as base_path:
dm = DependencyManager( default_base_path=base_path )
test_repo = __build_test_repo('package', version=TEST_VERSION)
ts_package_dir = __build_ts_test_package(base_path)
gx_env_path = __setup_galaxy_package_dep(base_path, TEST_REPO_NAME, TEST_VERSION)
ts_env_path = os.path.join(ts_package_dir, "env.sh")
dependency = dm.find_dep( TEST_REPO_NAME, version=TEST_VERSION, type='package', installed_tool_dependencies=[test_repo] )
assert dependency.script != gx_env_path # Not the galaxy path, it should be the tool shed path used.
assert dependency.script == ts_env_path
def __build_ts_test_package(base_path, script_contents=''):
package_dir = os.path.join(base_path, TEST_REPO_NAME, TEST_VERSION, TEST_REPO_USER, TEST_REPO_NAME, TEST_REPO_CHANGESET)
__touch(os.path.join(package_dir, 'env.sh'), script_contents)
return package_dir
def test_module_dependency_resolver():
with __test_base_path() as temp_directory:
module_script = os.path.join(temp_directory, "modulecmd")
__write_script(module_script, '''#!/bin/sh
cat %s/example_output 1>&2;
''' % temp_directory)
with open(os.path.join(temp_directory, "example_output"), "w") as f:
# Subset of module avail from MSI cluster.
f.write('''
-------------------------- /soft/modules/modulefiles ---------------------------
JAGS/3.2.0-gcc45
JAGS/3.3.0-gcc4.7.2
ProbABEL/0.1-3
ProbABEL/0.1-9e
R/2.12.2
R/2.13.1
R/2.14.1
R/2.15.0
R/2.15.1
R/3.0.1(default)
abokia-blast/2.0.2-130524/ompi_intel
abokia-blast/2.0.2-130630/ompi_intel
--------------------------- /soft/intel/modulefiles ----------------------------
advisor/2013/update1 intel/11.1.075 mkl/10.2.1.017
advisor/2013/update2 intel/11.1.080 mkl/10.2.5.035
advisor/2013/update3 intel/12.0 mkl/10.2.7.041
''')
resolver = ModuleDependencyResolver(None, modulecmd=module_script)
module = resolver.resolve( name="R", version=None, type="package" )
assert module.module_name == "R"
assert module.module_version == None
module = resolver.resolve( name="R", version="3.0.1", type="package" )
assert module.module_name == "R"
assert module.module_version == "3.0.1"
module = resolver.resolve( name="R", version="3.0.4", type="package" )
assert module == INDETERMINATE_DEPENDENCY
def test_module_dependency():
with __test_base_path() as temp_directory:
## Create mock modulecmd script that just exports a variable
## the way modulecmd sh load would, but also validate correct
## module name and version are coming through.
mock_modulecmd = os.path.join(temp_directory, 'modulecmd')
__write_script(mock_modulecmd, '''#!/bin/sh
if [ $3 != "foomodule/1.0" ];
then
exit 1
fi
echo 'FOO="bar"'
''')
resolver = Bunch(modulecmd=mock_modulecmd)
dependency = ModuleDependency(resolver, "foomodule", "1.0")
__assert_foo_exported( dependency.shell_commands( Bunch( type="package" ) ) )
def __write_script(path, contents):
with open(path, 'w') as f:
f.write(contents)
st = stat(path)
chmod(path, st.st_mode | S_IXUSR)
def test_galaxy_dependency_object_script():
with __test_base_path() as base_path:
## Create env.sh file that just exports variable FOO and verify it
## shell_commands export it correctly.
env_path = __setup_galaxy_package_dep(base_path, TEST_REPO_NAME, TEST_VERSION, "export FOO=\"bar\"")
dependency = GalaxyPackageDependency(env_path, os.path.dirname(env_path), TEST_VERSION)
__assert_foo_exported( dependency.shell_commands( Bunch( type="package" ) ) )
def test_shell_commands_built():
## Test that dependency manager builds valid shell commands for a list of
## requirements.
with __test_base_path() as base_path:
dm = DependencyManager( default_base_path=base_path )
__setup_galaxy_package_dep( base_path, TEST_REPO_NAME, TEST_VERSION, contents="export FOO=\"bar\"" )
mock_requirements = [ Bunch(type="package", version=TEST_VERSION, name=TEST_REPO_NAME ) ]
commands = dm.dependency_shell_commands( mock_requirements )
__assert_foo_exported( commands )
def __assert_foo_exported( commands ):
command = ["bash", "-c", "%s; echo \"$FOO\"" % "".join(commands)]
process = Popen(command, stdout=PIPE)
output = process.communicate()[0].strip()
assert output == 'bar', "Command %s exports FOO as %s, not bar" % (command, output)
def __setup_galaxy_package_dep(base_path, name, version, contents=""):
dep_directory = os.path.join( base_path, name, version )
env_path = os.path.join( dep_directory, "env.sh" )
__touch( env_path, contents )
return env_path
def __touch( fname, data=None ):
dirname = os.path.dirname( fname )
if not os.path.exists( dirname ):
makedirs( dirname )
f = open( fname, 'w' )
try:
if data:
f.write( data )
finally:
f.close()
def __build_test_repo(type, version=None):
return Bunch(
owner=TEST_REPO_USER,
name=TEST_REPO_NAME,
type=type,
version=version,
tool_shed_repository=Bunch(
owner=TEST_REPO_USER,
name=TEST_REPO_NAME,
installed_changeset_revision=TEST_REPO_CHANGESET
)
)
@contextmanager
def __test_base_path():
base_path = tempfile.mkdtemp()
try:
yield base_path
finally:
rmtree(base_path)
def test_parse():
with __parse_resolvers('''
''') as dependency_resolvers:
assert 'ToolShed' in dependency_resolvers[0].__class__.__name__
assert 'Galaxy' in dependency_resolvers[1].__class__.__name__
with __parse_resolvers('''
''') as dependency_resolvers:
assert 'Galaxy' in dependency_resolvers[0].__class__.__name__
assert 'ToolShed' in dependency_resolvers[1].__class__.__name__
with __parse_resolvers('''
''') as dependency_resolvers:
assert not dependency_resolvers[0].versionless
assert dependency_resolvers[2].versionless
with __parse_resolvers('''
''') as dependency_resolvers:
# Unspecified base_paths are both default_base_paths
assert dependency_resolvers[0].base_path == dependency_resolvers[1].base_path
# Can specify custom base path...
assert dependency_resolvers[2].base_path == "/opt/galaxy/legacy"
# ... that is different from the default.
assert dependency_resolvers[0].base_path != dependency_resolvers[2].base_path
def test_uses_tool_shed_dependencies():
with __dependency_manager('''
''') as dm:
assert not dm.uses_tool_shed_dependencies()
with __dependency_manager('''
''') as dm:
assert dm.uses_tool_shed_dependencies()
def test_config_module_defaults():
with __parse_resolvers('''
''') as dependency_resolvers:
module_resolver = dependency_resolvers[0]
assert module_resolver.module_checker.__class__.__name__ == "AvailModuleChecker"
def test_config_modulepath():
# Test reads and splits MODULEPATH if modulepath is not specified.
with __parse_resolvers('''
''') as dependency_resolvers:
assert dependency_resolvers[0].module_checker.directories == ["/opt/modules/modulefiles", "/usr/local/modules/modulefiles"]
def test_config_MODULEPATH():
# Test reads and splits MODULEPATH if modulepath is not specified.
with __environ({"MODULEPATH": "/opt/modules/modulefiles:/usr/local/modules/modulefiles"}):
with __parse_resolvers('''
''') as dependency_resolvers:
assert dependency_resolvers[0].module_checker.directories == ["/opt/modules/modulefiles", "/usr/local/modules/modulefiles"]
def test_config_MODULESHOME():
# Test fallbacks to read MODULESHOME if modulepath is not specified and
# neither is MODULEPATH.
with __environ({"MODULESHOME": "/opt/modules"}, remove="MODULEPATH"):
with __parse_resolvers('''
''') as dependency_resolvers:
assert dependency_resolvers[0].module_checker.directories == ["/opt/modules/modulefiles"]
def test_config_module_directory_searcher():
with __parse_resolvers('''
''') as dependency_resolvers:
module_resolver = dependency_resolvers[0]
assert module_resolver.module_checker.directories == ["/opt/Modules/modulefiles"]
@contextmanager
def __environ(values, remove=[]):
"""
Modify the environment for a test, adding/updating values in dict `values` and
removing any environment variables mentioned in list `remove`.
"""
new_keys = set(environ.keys()) - set(values.keys())
old_environ = environ.copy()
try:
environ.update(values)
for to_remove in remove:
try:
del environ[remove]
except KeyError:
pass
yield
finally:
environ.update(old_environ)
for key in new_keys:
del environ[key]
@contextmanager
def __parse_resolvers(xml_content):
with __dependency_manager(xml_content) as dm:
yield dm.dependency_resolvers
@contextmanager
def __dependency_manager(xml_content):
with __test_base_path() as base_path:
f = tempfile.NamedTemporaryFile()
f.write(xml_content)
f.flush()
dm = DependencyManager( default_base_path=base_path, conf_file=f.name )
yield dm