<s> import os <EOL> import re <EOL> import shutil <EOL> import time as t <EOL> from os . path import join as pjoin <EOL> from StringIO import StringIO <EOL> import tempfile <EOL> from nose . tools import assert_true , assert_equal <EOL> from numpy . testing import assert_array_equal <EOL> import smartdispatch <EOL> from smartdispatch import utils <EOL> def test_generate_name_from_command ( ) : <EOL> date_length = <NUM_LIT:20> <EOL> command = "<STR_LIT>" <EOL> expected = "<STR_LIT:_>" . join ( command . split ( ) ) <EOL> assert_equal ( smartdispatch . generate_name_from_command ( command ) [ date_length : ] , expected ) <EOL> max_length_arg = <NUM_LIT:7> <EOL> long_arg = "<STR_LIT>" <EOL> command = "<STR_LIT>" + long_arg + "<STR_LIT>" <EOL> expected = command . split ( ) <EOL> expected [ <NUM_LIT:1> ] = long_arg [ - max_length_arg : ] <EOL> expected = "<STR_LIT:_>" . join ( expected ) <EOL> assert_equal ( smartdispatch . generate_name_from_command ( command , max_length_arg ) [ date_length : ] , expected ) <EOL> max_length = <NUM_LIT> <EOL> command = "<STR_LIT>" <EOL> expected = command [ : max_length ] . replace ( "<STR_LIT:U+0020>" , "<STR_LIT:_>" ) <EOL> assert_equal ( smartdispatch . generate_name_from_command ( command , max_length = max_length + date_length ) [ date_length : ] , expected ) <EOL> command = "<STR_LIT>" <EOL> expected = "<STR_LIT>" <EOL> assert_equal ( smartdispatch . generate_name_from_command ( command ) [ date_length : ] , expected ) <EOL> def test_get_commands_from_file ( ) : <EOL> commands = [ "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ] <EOL> fileobj = StringIO ( "<STR_LIT:\n>" . join ( commands ) ) <EOL> assert_array_equal ( smartdispatch . get_commands_from_file ( fileobj ) , commands ) <EOL> fileobj = StringIO ( "<STR_LIT:\n>" . join ( commands ) + "<STR_LIT:\n>" ) <EOL> assert_array_equal ( smartdispatch . get_commands_from_file ( fileobj ) , commands ) <EOL> def test_unfold_command ( ) : <EOL> cmd = "<STR_LIT>" <EOL> assert_equal ( smartdispatch . unfold_command ( cmd ) , [ "<STR_LIT>" ] ) <EOL> cmd = "<STR_LIT>" <EOL> assert_equal ( smartdispatch . unfold_command ( cmd ) , [ "<STR_LIT>" ] ) <EOL> cmd = "<STR_LIT>" <EOL> assert_equal ( smartdispatch . unfold_command ( cmd ) , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> cmd = "<STR_LIT>" <EOL> assert_equal ( smartdispatch . unfold_command ( cmd ) , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> cmd = "<STR_LIT>" <EOL> assert_equal ( smartdispatch . unfold_command ( cmd ) , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> cmd = "<STR_LIT>" <EOL> assert_equal ( smartdispatch . unfold_command ( cmd ) , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> cmd = "<STR_LIT>" <EOL> assert_equal ( smartdispatch . unfold_command ( cmd ) , [ "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ] ) <EOL> cmd = "<STR_LIT>" <EOL> assert_equal ( smartdispatch . unfold_command ( cmd ) , [ "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ] ) <EOL> cmd = "<STR_LIT>" <EOL> assert_equal ( smartdispatch . unfold_command ( cmd ) , [ "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ] ) <EOL> cmd = "<STR_LIT>" <EOL> assert_equal ( smartdispatch . unfold_command ( cmd ) , [ "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ] ) <EOL> def test_replace_uid_tag ( ) : <EOL> command = "<STR_LIT>" <EOL> assert_array_equal ( smartdispatch . replace_uid_tag ( [ command ] ) , [ command ] ) <EOL> command = "<STR_LIT>" <EOL> uid = utils . generate_uid_from_string ( command ) <EOL> assert_array_equal ( smartdispatch . replace_uid_tag ( [ command ] ) , [ command . replace ( "<STR_LIT>" , uid ) ] ) <EOL> command = "<STR_LIT>" <EOL> uid = utils . generate_uid_from_string ( command ) <EOL> assert_array_equal ( smartdispatch . replace_uid_tag ( [ command ] ) , [ command . replace ( "<STR_LIT>" , uid ) ] ) <EOL> commands = [ "<STR_LIT>" ] * <NUM_LIT:10> <EOL> uid = utils . generate_uid_from_string ( commands [ <NUM_LIT:0> ] ) <EOL> assert_array_equal ( smartdispatch . replace_uid_tag ( commands ) , [ commands [ <NUM_LIT:0> ] . replace ( "<STR_LIT>" , uid ) ] * len ( commands ) ) <EOL> def test_get_available_queues ( ) : <EOL> assert_equal ( smartdispatch . get_available_queues ( cluster_name = None ) , { } ) <EOL> assert_equal ( smartdispatch . get_available_queues ( cluster_name = "<STR_LIT>" ) , { } ) <EOL> queues_infos = smartdispatch . get_available_queues ( cluster_name = "<STR_LIT>" ) <EOL> assert_true ( len ( queues_infos ) > <NUM_LIT:0> ) <EOL> queues_infos = smartdispatch . get_available_queues ( cluster_name = "<STR_LIT>" ) <EOL> assert_true ( len ( queues_infos ) > <NUM_LIT:0> ) <EOL> def test_get_job_folders ( ) : <EOL> temp_dir = tempfile . mkdtemp ( ) <EOL> jobname = "<STR_LIT>" <EOL> job_folders_paths = smartdispatch . get_job_folders ( temp_dir , jobname ) <EOL> path_job , path_job_logs , path_job_commands = job_folders_paths <EOL> assert_true ( jobname in path_job ) <EOL> assert_true ( os . path . isdir ( path_job ) ) <EOL> assert_equal ( os . path . basename ( path_job ) , jobname ) <EOL> assert_true ( jobname in path_job_logs ) <EOL> assert_true ( os . path . isdir ( path_job_logs ) ) <EOL> assert_true ( os . path . isdir ( pjoin ( path_job_logs , '<STR_LIT>' ) ) ) <EOL> assert_true ( os . path . isdir ( pjoin ( path_job_logs , '<STR_LIT>' ) ) ) <EOL> assert_true ( os . path . isdir ( path_job_logs ) ) <EOL> assert_equal ( os . path . basename ( path_job_logs ) , "<STR_LIT>" ) <EOL> assert_true ( jobname in path_job_commands ) <EOL> assert_true ( os . path . isdir ( path_job_commands ) ) <EOL> assert_equal ( os . path . basename ( path_job_commands ) , "<STR_LIT>" ) <EOL> jobname += "<STR_LIT:2>" <EOL> os . rename ( path_job , path_job + "<STR_LIT:2>" ) <EOL> job_folders_paths = smartdispatch . get_job_folders ( temp_dir , jobname ) <EOL> path_job , path_job_logs , path_job_commands = job_folders_paths <EOL> assert_true ( jobname in path_job ) <EOL> assert_true ( os . path . isdir ( path_job ) ) <EOL> assert_equal ( os . path . basename ( path_job ) , jobname ) <EOL> assert_true ( jobname in path_job_logs ) <EOL> assert_true ( os . path . isdir ( path_job_logs ) ) <EOL> assert_true ( os . path . isdir ( pjoin ( path_job_logs , '<STR_LIT>' ) ) ) <EOL> assert_true ( os . path . isdir ( pjoin ( path_job_logs , '<STR_LIT>' ) ) ) <EOL> assert_true ( os . path . isdir ( path_job_logs ) ) <EOL> assert_equal ( os . path . basename ( path_job_logs ) , "<STR_LIT>" ) <EOL> assert_true ( jobname in path_job_commands ) <EOL> assert_true ( os . path . isdir ( path_job_commands ) ) <EOL> assert_equal ( os . path . basename ( path_job_commands ) , "<STR_LIT>" ) <EOL> shutil . rmtree ( temp_dir ) <EOL> def test_log_command_line ( ) : <EOL> temp_dir = tempfile . mkdtemp ( ) <EOL> command_line_log_file = pjoin ( temp_dir , "<STR_LIT>" ) <EOL> command_1 = "<STR_LIT>" <EOL> smartdispatch . log_command_line ( temp_dir , command_1 ) <EOL> assert_true ( os . path . isfile ( command_line_log_file ) ) <EOL> lines = open ( command_line_log_file ) . read ( ) . strip ( ) . split ( "<STR_LIT:\n>" ) <EOL> assert_equal ( len ( lines ) , <NUM_LIT:2> ) <EOL> assert_true ( t . strftime ( "<STR_LIT>" ) in lines [ <NUM_LIT:0> ] ) <EOL> assert_equal ( lines [ <NUM_LIT:1> ] , command_1 ) <EOL> command_2 = "<STR_LIT>" <EOL> smartdispatch . log_command_line ( temp_dir , command_2 ) <EOL> assert_true ( os . path . isfile ( command_line_log_file ) ) <EOL> lines = open ( command_line_log_file ) . read ( ) . strip ( ) . split ( "<STR_LIT:\n>" ) <EOL> assert_equal ( len ( lines ) , <NUM_LIT:5> ) <EOL> assert_true ( t . strftime ( "<STR_LIT>" ) in lines [ <NUM_LIT:3> ] ) <EOL> assert_equal ( lines [ <NUM_LIT:4> ] , command_2 . replace ( '<STR_LIT:">' , r'<STR_LIT>' ) ) <EOL> command_3 = "<STR_LIT>" <EOL> smartdispatch . log_command_line ( temp_dir , command_3 ) <EOL> assert_true ( os . path . isfile ( command_line_log_file ) ) <EOL> lines = open ( command_line_log_file ) . read ( ) . strip ( ) . split ( "<STR_LIT:\n>" ) <EOL> assert_equal ( len ( lines ) , <NUM_LIT:8> ) <EOL> assert_true ( t . strftime ( "<STR_LIT>" ) in lines [ <NUM_LIT:6> ] ) <EOL> assert_equal ( lines [ <NUM_LIT:7> ] , re . sub ( r'<STR_LIT>' , r'<STR_LIT>' , command_3 ) ) <EOL> shutil . rmtree ( temp_dir ) </s>
