<s> from collections import deque <EOL> import os <EOL> from . test_utils import TempDirectoryTestCase <EOL> from pulsar . client . test . test_common import write_config <EOL> from pulsar . client import submit_job , ClientJobDescription <EOL> from pulsar . client import ClientOutputs <EOL> from galaxy . tools . deps . dependencies import DependenciesDescription <EOL> from galaxy . tools . deps . requirements import ToolRequirement <EOL> TEST_REQUIREMENT_1 = ToolRequirement ( "<STR_LIT>" , "<STR_LIT:1.0>" ) <EOL> TEST_REQUIREMENT_2 = ToolRequirement ( "<STR_LIT>" , "<STR_LIT:1.0>" ) <EOL> TEST_ENV_1 = dict ( name = "<STR_LIT:x>" , value = "<STR_LIT:y>" ) <EOL> class TestStager ( TempDirectoryTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestStager , self ) . setUp ( ) <EOL> from . test_utils import get_test_tool <EOL> self . tool = get_test_tool ( ) <EOL> self . client = MockClient ( self . temp_directory , self . tool ) <EOL> inputs = self . __setup_inputs ( ) <EOL> self . client_job_description = ClientJobDescription ( <EOL> tool = self . tool , <EOL> command_line = "<STR_LIT>" , <EOL> config_files = [ ] , <EOL> input_files = inputs , <EOL> client_outputs = ClientOutputs ( "<STR_LIT>" , [ ] ) , <EOL> working_directory = "<STR_LIT>" , <EOL> dependencies_description = DependenciesDescription ( requirements = [ TEST_REQUIREMENT_1 , TEST_REQUIREMENT_2 ] ) , <EOL> env = [ TEST_ENV_1 ] , <EOL> rewrite_paths = False , <EOL> ) <EOL> self . job_config = dict ( <EOL> working_directory = "<STR_LIT>" , <EOL> outputs_directory = "<STR_LIT>" , <EOL> system_properties = dict ( <EOL> separator = "<STR_LIT:\\>" , <EOL> ) , <EOL> ) <EOL> def __setup_inputs ( self ) : <EOL> files_directory = os . path . join ( self . temp_directory , "<STR_LIT>" ) <EOL> os . makedirs ( files_directory ) <EOL> self . input1 = os . path . join ( files_directory , "<STR_LIT>" ) <EOL> self . input1_files_path = os . path . join ( files_directory , "<STR_LIT>" ) <EOL> open ( self . input1 , "<STR_LIT:wb>" ) . write ( b"<STR_LIT>" ) <EOL> self . input2 = os . path . join ( files_directory , "<STR_LIT>" ) <EOL> open ( self . input2 , "<STR_LIT:wb>" ) . write ( b"<STR_LIT>" ) <EOL> return [ self . input1 , self . input2 ] <EOL> def test_tool_file_rewrite ( self ) : <EOL> self . client_job_description . rewrite_paths = True <EOL> tool_dir = os . path . abspath ( self . tool . tool_dir ) <EOL> command_line = "<STR_LIT>" % tool_dir <EOL> self . client_job_description . command_line = command_line <EOL> rewritten_command_line = "<STR_LIT>" <EOL> self . client . expect_put_paths ( [ "<STR_LIT>" ] ) <EOL> self . client . expect_command_line ( rewritten_command_line ) <EOL> self . _submit ( ) <EOL> uploaded_file1 = self . client . put_files [ <NUM_LIT:0> ] <EOL> assert uploaded_file1 [ <NUM_LIT:1> ] == "<STR_LIT>" <EOL> self . assertEquals ( uploaded_file1 [ <NUM_LIT:0> ] , "<STR_LIT>" % tool_dir ) <EOL> def test_input_extra_rewrite ( self ) : <EOL> self . client_job_description . rewrite_paths = True <EOL> extra_file = os . path . join ( self . input1_files_path , "<STR_LIT>" , "<STR_LIT>" ) <EOL> os . makedirs ( os . path . dirname ( extra_file ) ) <EOL> open ( extra_file , "<STR_LIT:w>" ) . write ( "<STR_LIT>" ) <EOL> command_line = "<STR_LIT>" % extra_file <EOL> self . client_job_description . command_line = command_line <EOL> self . client . expect_command_line ( "<STR_LIT>" ) <EOL> self . client . expect_put_paths ( [ "<STR_LIT>" ] ) <EOL> self . _submit ( ) <EOL> uploaded_file1 = self . client . put_files [ <NUM_LIT:0> ] <EOL> assert uploaded_file1 [ <NUM_LIT:1> ] == "<STR_LIT:input>" <EOL> assert uploaded_file1 [ <NUM_LIT:0> ] == extra_file <EOL> def test_unstructured_rewrite ( self ) : <EOL> self . client_job_description . rewrite_paths = True <EOL> self . client . set_action_map_config ( dict ( paths = [ <EOL> dict ( path = self . temp_directory , path_types = "<STR_LIT>" ) <EOL> ] ) ) <EOL> local_unstructured_file = os . path . join ( self . temp_directory , "<STR_LIT>" ) <EOL> open ( local_unstructured_file , "<STR_LIT:wb>" ) . write ( b"<STR_LIT>" ) <EOL> command_line = "<STR_LIT>" % local_unstructured_file <EOL> self . client_job_description . command_line = command_line <EOL> self . client . expect_put_paths ( [ "<STR_LIT>" ] ) <EOL> self . client . expect_command_line ( "<STR_LIT>" ) <EOL> self . _submit ( ) <EOL> uploaded_file1 = self . client . put_files [ <NUM_LIT:0> ] <EOL> assert uploaded_file1 [ <NUM_LIT:1> ] == "<STR_LIT>" <EOL> self . assertEquals ( uploaded_file1 [ <NUM_LIT:0> ] , local_unstructured_file ) <EOL> def test_submit_no_rewrite ( self ) : <EOL> command_line_template = "<STR_LIT>" <EOL> command_line = command_line_template % ( self . input1 , self . input2 ) <EOL> self . client_job_description . command_line = command_line <EOL> self . client . expect_command_line ( command_line ) <EOL> self . _submit ( ) <EOL> self . _assert_inputs_uploaded ( ) <EOL> def test_submit_rewrite ( self ) : <EOL> self . client_job_description . rewrite_paths = True <EOL> command_line_template = "<STR_LIT>" <EOL> self . client_job_description . command_line = command_line_template % ( self . input1 , self . input2 ) <EOL> rewritten_paths = ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ) <EOL> self . client . expect_command_line ( command_line_template % rewritten_paths ) <EOL> self . _submit ( ) <EOL> self . _assert_inputs_uploaded ( ) <EOL> def _assert_inputs_uploaded ( self ) : <EOL> uploaded_file1 = self . client . put_files [ <NUM_LIT:0> ] <EOL> assert uploaded_file1 [ <NUM_LIT:1> ] == "<STR_LIT:input>" <EOL> assert uploaded_file1 [ <NUM_LIT:0> ] == self . input1 <EOL> uploaded_file2 = self . client . put_files [ <NUM_LIT:1> ] <EOL> assert uploaded_file2 [ <NUM_LIT:1> ] == "<STR_LIT:input>" <EOL> assert uploaded_file2 [ <NUM_LIT:0> ] == self . input2 <EOL> def _submit ( self ) : <EOL> return submit_job ( self . client , self . client_job_description , self . job_config ) <EOL> class MockClient ( object ) : <EOL> def __init__ ( self , temp_directory , tool ) : <EOL> self . temp_directory = temp_directory <EOL> self . default_file_action = "<STR_LIT>" <EOL> self . action_config_path = None <EOL> self . files_endpoint = None <EOL> self . expected_tool = tool <EOL> self . job_id = "<STR_LIT>" <EOL> self . expected_command_line = None <EOL> self . expect_put_paths ( [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ] ) <EOL> self . put_files = [ ] <EOL> def set_action_map_config ( self , config ) : <EOL> self . action_config_path = write_config ( self , config , name = "<STR_LIT>" ) <EOL> def expect_put_paths ( self , paths ) : <EOL> self . put_paths = deque ( paths ) <EOL> def setup ( self , tool_id , tool_version , use_metadata = False ) : <EOL> assert tool_id == self . expected_tool . id <EOL> assert tool_version == self . expected_tool . version <EOL> return { } <EOL> def launch ( self , command_line , dependencies_description , job_config = { } , remote_staging = { } , env = [ ] ) : <EOL> if self . expected_command_line is not None : <EOL> message = "<STR_LIT>" % ( self . expected_command_line , command_line ) <EOL> assert self . expected_command_line == command_line , message <EOL> assert dependencies_description . requirements == [ TEST_REQUIREMENT_1 , TEST_REQUIREMENT_2 ] <EOL> assert env == [ TEST_ENV_1 ] <EOL> def expect_command_line ( self , expected_command_line ) : <EOL> self . expected_command_line = expected_command_line <EOL> def put_file ( self , path , type , name , contents , action_type = '<STR_LIT>' ) : <EOL> self . put_files . append ( ( path , type , name , contents ) ) <EOL> return { "<STR_LIT:path>" : self . put_paths . popleft ( ) } </s>
