mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-10-25 00:12:08 +08:00
@@ -0,0 +1,74 @@
|
||||
from os import getcwd
|
||||
from os.path import abspath
|
||||
|
||||
|
||||
def build_command( job, job_wrapper, include_metadata=False, include_work_dir_outputs=True ):
|
||||
"""
|
||||
Compose the sequence of commands necessary to execute a job. This will
|
||||
currently include:
|
||||
|
||||
- environment settings corresponding to any requirement tags
|
||||
- preparing input files
|
||||
- command line taken from job wrapper
|
||||
- commands to set metadata (if include_metadata is True)
|
||||
"""
|
||||
|
||||
commands = job_wrapper.get_command_line()
|
||||
|
||||
# All job runners currently handle this case which should never occur
|
||||
if not commands:
|
||||
return None
|
||||
|
||||
# Remove trailing semi-colon so we can start hacking up this command.
|
||||
# TODO: Refactor to compose a list and join with ';', would be more clean.
|
||||
commands = commands.rstrip(";")
|
||||
|
||||
# Prepend version string
|
||||
if job_wrapper.version_string_cmd:
|
||||
commands = "%s &> %s; " % ( job_wrapper.version_string_cmd, job_wrapper.get_version_string_path() ) + commands
|
||||
|
||||
# prepend getting input files (if defined)
|
||||
if hasattr(job_wrapper, 'prepare_input_files_cmds') and job_wrapper.prepare_input_files_cmds is not None:
|
||||
commands = "; ".join( job_wrapper.prepare_input_files_cmds + [ commands ] )
|
||||
|
||||
# Prepend dependency injection
|
||||
if job_wrapper.dependency_shell_commands:
|
||||
commands = "; ".join( job_wrapper.dependency_shell_commands + [ commands ] )
|
||||
|
||||
# Coping work dir outputs or setting metadata will mask return code of
|
||||
# tool command. If these are used capture the return code and ensure
|
||||
# the last thing that happens is an exit with return code.
|
||||
capture_return_code_command = "; return_code=$?"
|
||||
captured_return_code = False
|
||||
|
||||
# Append commands to copy job outputs based on from_work_dir attribute.
|
||||
if include_work_dir_outputs:
|
||||
work_dir_outputs = job.get_work_dir_outputs( job_wrapper )
|
||||
if work_dir_outputs:
|
||||
if not captured_return_code:
|
||||
commands += capture_return_code_command
|
||||
captured_return_code = True
|
||||
|
||||
commands += "; " + "; ".join( [ "if [ -f %s ] ; then cp %s %s ; fi" %
|
||||
( source_file, source_file, destination ) for ( source_file, destination ) in work_dir_outputs ] )
|
||||
|
||||
# Append metadata setting commands, we don't want to overwrite metadata
|
||||
# that was copied over in init_meta(), as per established behavior
|
||||
if include_metadata and job_wrapper.requires_setting_metadata:
|
||||
if not captured_return_code:
|
||||
commands += capture_return_code_command
|
||||
captured_return_code = True
|
||||
commands += "; cd %s; " % abspath( getcwd() )
|
||||
commands += job_wrapper.setup_external_metadata(
|
||||
exec_dir=abspath( getcwd() ),
|
||||
tmp_dir=job_wrapper.working_directory,
|
||||
dataset_files_path=job.app.model.Dataset.file_path,
|
||||
output_fnames=job_wrapper.get_output_fnames(),
|
||||
set_extension=False,
|
||||
kwds={ 'overwrite' : False }
|
||||
)
|
||||
|
||||
if captured_return_code:
|
||||
commands += '; sh -c "exit $return_code"'
|
||||
|
||||
return commands
|
||||
@@ -12,6 +12,7 @@ import subprocess
|
||||
from Queue import Queue, Empty
|
||||
|
||||
import galaxy.jobs
|
||||
from galaxy.jobs.command_factory import build_command
|
||||
from galaxy import model
|
||||
from galaxy.util import DATABASE_MAX_STRING_SIZE, shrink_stream_by_size
|
||||
|
||||
@@ -142,67 +143,7 @@ class BaseJobRunner( object ):
|
||||
raise NotImplementedError()
|
||||
|
||||
def build_command_line( self, job_wrapper, include_metadata=False, include_work_dir_outputs=True ):
|
||||
"""
|
||||
Compose the sequence of commands necessary to execute a job. This will
|
||||
currently include:
|
||||
|
||||
- environment settings corresponding to any requirement tags
|
||||
- preparing input files
|
||||
- command line taken from job wrapper
|
||||
- commands to set metadata (if include_metadata is True)
|
||||
"""
|
||||
|
||||
commands = job_wrapper.get_command_line()
|
||||
# All job runners currently handle this case which should never
|
||||
# occur
|
||||
if not commands:
|
||||
return None
|
||||
# Prepend version string
|
||||
if job_wrapper.version_string_cmd:
|
||||
commands = "%s &> %s; " % ( job_wrapper.version_string_cmd, job_wrapper.get_version_string_path() ) + commands
|
||||
# prepend getting input files (if defined)
|
||||
if hasattr(job_wrapper, 'prepare_input_files_cmds') and job_wrapper.prepare_input_files_cmds is not None:
|
||||
commands = "; ".join( job_wrapper.prepare_input_files_cmds + [ commands ] )
|
||||
# Prepend dependency injection
|
||||
if job_wrapper.dependency_shell_commands:
|
||||
commands = "; ".join( job_wrapper.dependency_shell_commands + [ commands ] )
|
||||
|
||||
# Coping work dir outputs or setting metadata will mask return code of
|
||||
# tool command. If these are used capture the return code and ensure
|
||||
# the last thing that happens is an exit with return code.
|
||||
capture_return_code_command = "; return_code=$?"
|
||||
captured_return_code = False
|
||||
|
||||
# Append commands to copy job outputs based on from_work_dir attribute.
|
||||
if include_work_dir_outputs:
|
||||
work_dir_outputs = self.get_work_dir_outputs( job_wrapper )
|
||||
if work_dir_outputs:
|
||||
if not captured_return_code:
|
||||
commands += capture_return_code_command
|
||||
captured_return_code = True
|
||||
commands += "; " + "; ".join( [ "if [ -f %s ] ; then cp %s %s ; fi" %
|
||||
( source_file, source_file, destination ) for ( source_file, destination ) in work_dir_outputs ] )
|
||||
|
||||
# Append metadata setting commands, we don't want to overwrite metadata
|
||||
# that was copied over in init_meta(), as per established behavior
|
||||
if include_metadata and job_wrapper.requires_setting_metadata:
|
||||
if not captured_return_code:
|
||||
commands += capture_return_code_command
|
||||
captured_return_code = True
|
||||
commands += "; cd %s; " % os.path.abspath( os.getcwd() )
|
||||
commands += job_wrapper.setup_external_metadata(
|
||||
exec_dir = os.path.abspath( os.getcwd() ),
|
||||
tmp_dir = job_wrapper.working_directory,
|
||||
dataset_files_path = self.app.model.Dataset.file_path,
|
||||
output_fnames = job_wrapper.get_output_fnames(),
|
||||
set_extension = False,
|
||||
kwds = { 'overwrite' : False } )
|
||||
|
||||
|
||||
if captured_return_code:
|
||||
commands += '; sh -c "exit $return_code"'
|
||||
|
||||
return commands
|
||||
return build_command( self, job_wrapper, include_metadata=include_metadata, include_work_dir_outputs=include_work_dir_outputs )
|
||||
|
||||
def get_work_dir_outputs( self, job_wrapper ):
|
||||
"""
|
||||
|
||||
@@ -0,0 +1,86 @@
|
||||
from os import getcwd
|
||||
from unittest import TestCase
|
||||
|
||||
from galaxy.jobs.command_factory import build_command
|
||||
from galaxy.util.bunch import Bunch
|
||||
|
||||
MOCK_COMMAND_LINE = "/opt/galaxy/tools/bowtie /mnt/galaxyData/files/000/input000.dat"
|
||||
|
||||
|
||||
class TestCommandFactory(TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.job_wrapper = MockJobWrapper()
|
||||
self.job = Bunch(app=Bunch(model=Bunch(Dataset=Bunch(file_path="file_path"))))
|
||||
self.include_metadata = False
|
||||
self.include_work_dir_outputs = True
|
||||
|
||||
def test_simplest_command(self):
|
||||
self.include_work_dir_outputs = False
|
||||
self.__assert_command_is( MOCK_COMMAND_LINE )
|
||||
|
||||
def test_shell_commands(self):
|
||||
self.include_work_dir_outputs = False
|
||||
dep_commands = [". /opt/galaxy/tools/bowtie/default/env.sh"]
|
||||
self.job_wrapper.dependency_shell_commands = dep_commands
|
||||
self.__assert_command_is( "%s; %s" % (dep_commands[0], MOCK_COMMAND_LINE) )
|
||||
|
||||
def test_set_metadata_skipped_if_unneeded(self):
|
||||
self.include_metadata = True
|
||||
self.include_work_dir_outputs = False
|
||||
self.__assert_command_is( MOCK_COMMAND_LINE )
|
||||
|
||||
def test_set_metadata(self):
|
||||
self._test_set_metadata()
|
||||
|
||||
def test_strips_trailing_semicolons(self):
|
||||
self.job_wrapper.command_line = "%s;" % MOCK_COMMAND_LINE
|
||||
self._test_set_metadata()
|
||||
|
||||
def _test_set_metadata(self):
|
||||
self.include_metadata = True
|
||||
self.include_work_dir_outputs = False
|
||||
metadata_line = "set_metadata_and_stuff.sh"
|
||||
self.job_wrapper.metadata_line = metadata_line
|
||||
expected_command = '%s; return_code=$?; cd %s; %s; sh -c "exit $return_code"' % (MOCK_COMMAND_LINE, getcwd(), metadata_line)
|
||||
self.__assert_command_is( expected_command )
|
||||
|
||||
def __assert_command_is(self, expected_command):
|
||||
command = self.__command()
|
||||
self.assertEqual(command, expected_command)
|
||||
|
||||
def __command(self):
|
||||
kwds = dict(
|
||||
job=self.job,
|
||||
job_wrapper=self.job_wrapper,
|
||||
include_metadata=self.include_metadata,
|
||||
include_work_dir_outputs=self.include_work_dir_outputs,
|
||||
)
|
||||
return build_command(**kwds)
|
||||
|
||||
|
||||
class MockJobWrapper(object):
|
||||
|
||||
def __init__(self):
|
||||
self.version_string_cmd = None
|
||||
self.command_line = MOCK_COMMAND_LINE
|
||||
self.dependency_shell_commands = []
|
||||
self.metadata_line = None
|
||||
self.working_directory = "job1"
|
||||
|
||||
def get_command_line(self):
|
||||
return self.command_line
|
||||
|
||||
@property
|
||||
def requires_setting_metadata(self):
|
||||
return self.metadata_line is not None
|
||||
|
||||
def setup_external_metadata(self, *args, **kwds):
|
||||
return self.metadata_line
|
||||
|
||||
def get_output_fnames(self):
|
||||
return []
|
||||
|
||||
|
||||
class MockJob(object):
|
||||
app = Bunch()
|
||||
Reference in New Issue
Block a user