<s> from __future__ import with_statement <EOL> from contextlib import nested <EOL> from StringIO import StringIO <EOL> from nose . tools import ok_ <EOL> from fudge import with_fakes , Fake <EOL> from fudge . inspector import arg as fudge_arg <EOL> from paramiko . sftp_client import SFTPClient <EOL> from fabric . state import env , output <EOL> from fabric . operations import require , prompt , _sudo_prefix , _shell_wrap , _shell_escape <EOL> from fabric . api import get , put , hide , show , cd , lcd , local , run , sudo , quiet <EOL> from fabric . exceptions import CommandTimeout <EOL> from fabric . decorators import with_settings <EOL> from utils import * <EOL> from server import ( server , PORT , RESPONSES , FILES , PASSWORDS , CLIENT_PRIVKEY , <EOL> USER , CLIENT_PRIVKEY_PASSPHRASE ) <EOL> def test_require_single_existing_key ( ) : <EOL> """<STR_LIT>""" <EOL> require ( '<STR_LIT:version>' ) <EOL> def test_require_multiple_existing_keys ( ) : <EOL> """<STR_LIT>""" <EOL> require ( '<STR_LIT:version>' , '<STR_LIT>' ) <EOL> @ aborts <EOL> def test_require_single_missing_key ( ) : <EOL> """<STR_LIT>""" <EOL> require ( '<STR_LIT>' ) <EOL> @ aborts <EOL> def test_require_multiple_missing_keys ( ) : <EOL> """<STR_LIT>""" <EOL> require ( '<STR_LIT:foo>' , '<STR_LIT:bar>' ) <EOL> @ aborts <EOL> def test_require_mixed_state_keys ( ) : <EOL> """<STR_LIT>""" <EOL> require ( '<STR_LIT:foo>' , '<STR_LIT:version>' ) <EOL> @ mock_streams ( '<STR_LIT>' ) <EOL> def test_require_mixed_state_keys_prints_missing_only ( ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> require ( '<STR_LIT:foo>' , '<STR_LIT:version>' ) <EOL> except SystemExit : <EOL> err = sys . stderr . getvalue ( ) <EOL> assert '<STR_LIT:version>' not in err <EOL> assert '<STR_LIT:foo>' in err <EOL> @ aborts <EOL> def test_require_iterable_provided_by_key ( ) : <EOL> """<STR_LIT>""" <EOL> def fake_providing_function ( ) : <EOL> pass <EOL> require ( '<STR_LIT:foo>' , provided_by = [ fake_providing_function ] ) <EOL> @ aborts <EOL> def test_require_noniterable_provided_by_key ( ) : <EOL> """<STR_LIT>""" <EOL> def fake_providing_function ( ) : <EOL> pass <EOL> require ( '<STR_LIT:foo>' , provided_by = fake_providing_function ) <EOL> @ aborts <EOL> def test_require_key_exists_empty_list ( ) : <EOL> """<STR_LIT>""" <EOL> require ( '<STR_LIT>' ) <EOL> @ aborts <EOL> @ with_settings ( foo = { } ) <EOL> def test_require_key_exists_empty_dict ( ) : <EOL> """<STR_LIT>""" <EOL> require ( '<STR_LIT:foo>' ) <EOL> @ aborts <EOL> @ with_settings ( foo = ( ) ) <EOL> def test_require_key_exists_empty_tuple ( ) : <EOL> """<STR_LIT>""" <EOL> require ( '<STR_LIT:foo>' ) <EOL> @ aborts <EOL> @ with_settings ( foo = set ( ) ) <EOL> def test_require_key_exists_empty_set ( ) : <EOL> """<STR_LIT>""" <EOL> require ( '<STR_LIT:foo>' ) <EOL> @ with_settings ( foo = <NUM_LIT:0> , bar = False ) <EOL> def test_require_key_exists_false_primitive_values ( ) : <EOL> """<STR_LIT>""" <EOL> require ( '<STR_LIT:foo>' , '<STR_LIT:bar>' ) <EOL> @ with_settings ( foo = [ '<STR_LIT:foo>' ] , bar = { '<STR_LIT:bar>' : '<STR_LIT:bar>' } , baz = ( '<STR_LIT>' , ) , qux = set ( '<STR_LIT>' ) ) <EOL> def test_require_complex_non_empty_values ( ) : <EOL> """<STR_LIT>""" <EOL> require ( '<STR_LIT:foo>' , '<STR_LIT:bar>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def p ( x ) : <EOL> sys . stdout . write ( x ) <EOL> @ mock_streams ( '<STR_LIT>' ) <EOL> @ with_patched_input ( p ) <EOL> def test_prompt_appends_space ( ) : <EOL> """<STR_LIT>""" <EOL> s = "<STR_LIT>" <EOL> prompt ( s ) <EOL> eq_ ( sys . stdout . getvalue ( ) , s + '<STR_LIT:U+0020>' ) <EOL> @ mock_streams ( '<STR_LIT>' ) <EOL> @ with_patched_input ( p ) <EOL> def test_prompt_with_default ( ) : <EOL> """<STR_LIT>""" <EOL> s = "<STR_LIT>" <EOL> d = "<STR_LIT>" <EOL> prompt ( s , default = d ) <EOL> eq_ ( sys . stdout . getvalue ( ) , "<STR_LIT>" % ( s , d ) ) <EOL> def test_sudo_prefix_with_user ( ) : <EOL> """<STR_LIT>""" <EOL> eq_ ( <EOL> _sudo_prefix ( user = "<STR_LIT:foo>" , group = None ) , <EOL> "<STR_LIT>" % ( env . sudo_prefix % env ) <EOL> ) <EOL> def test_sudo_prefix_without_user ( ) : <EOL> """<STR_LIT>""" <EOL> eq_ ( _sudo_prefix ( user = None , group = None ) , env . sudo_prefix % env ) <EOL> def test_sudo_prefix_with_group ( ) : <EOL> """<STR_LIT>""" <EOL> eq_ ( <EOL> _sudo_prefix ( user = None , group = "<STR_LIT:foo>" ) , <EOL> "<STR_LIT>" % ( env . sudo_prefix % env ) <EOL> ) <EOL> def test_sudo_prefix_with_user_and_group ( ) : <EOL> """<STR_LIT>""" <EOL> eq_ ( <EOL> _sudo_prefix ( user = "<STR_LIT:foo>" , group = "<STR_LIT:bar>" ) , <EOL> "<STR_LIT>" % ( env . sudo_prefix % env ) <EOL> ) <EOL> @ with_settings ( use_shell = True ) <EOL> def test_shell_wrap ( ) : <EOL> prefix = "<STR_LIT>" <EOL> command = "<STR_LIT>" <EOL> for description , shell , sudo_prefix , result in ( <EOL> ( "<STR_LIT>" , <EOL> True , None , '<STR_LIT>' % ( env . shell , command ) ) , <EOL> ( "<STR_LIT>" , <EOL> True , prefix , prefix + '<STR_LIT>' % ( env . shell , command ) ) , <EOL> ( "<STR_LIT>" , <EOL> False , None , command ) , <EOL> ( "<STR_LIT>" , <EOL> False , prefix , prefix + "<STR_LIT:U+0020>" + command ) , <EOL> ) : <EOL> eq_ . description = "<STR_LIT>" % description <EOL> yield eq_ , _shell_wrap ( command , shell_escape = True , shell = shell , sudo_prefix = sudo_prefix ) , result <EOL> del eq_ . description <EOL> @ with_settings ( use_shell = True ) <EOL> def test_shell_wrap_escapes_command_if_shell_is_true ( ) : <EOL> """<STR_LIT>""" <EOL> cmd = "<STR_LIT>" <EOL> eq_ ( <EOL> _shell_wrap ( cmd , shell_escape = True , shell = True ) , <EOL> '<STR_LIT>' % ( env . shell , _shell_escape ( cmd ) ) <EOL> ) <EOL> @ with_settings ( use_shell = True ) <EOL> def test_shell_wrap_does_not_escape_command_if_shell_is_true_and_shell_escape_is_false ( ) : <EOL> """<STR_LIT>""" <EOL> cmd = "<STR_LIT>" <EOL> eq_ ( <EOL> _shell_wrap ( cmd , shell_escape = False , shell = True ) , <EOL> '<STR_LIT>' % ( env . shell , cmd ) <EOL> ) <EOL> def test_shell_wrap_does_not_escape_command_if_shell_is_false ( ) : <EOL> """<STR_LIT>""" <EOL> cmd = "<STR_LIT>" <EOL> eq_ ( _shell_wrap ( cmd , shell_escape = True , shell = False ) , cmd ) <EOL> def test_shell_escape_escapes_doublequotes ( ) : <EOL> """<STR_LIT>""" <EOL> cmd = "<STR_LIT>" <EOL> eq_ ( _shell_escape ( cmd ) , '<STR_LIT>' ) <EOL> def test_shell_escape_escapes_dollar_signs ( ) : <EOL> """<STR_LIT>""" <EOL> cmd = "<STR_LIT>" <EOL> eq_ ( _shell_escape ( cmd ) , '<STR_LIT>' ) <EOL> def test_shell_escape_escapes_backticks ( ) : <EOL> """<STR_LIT>""" <EOL> cmd = "<STR_LIT>" <EOL> eq_ ( _shell_escape ( cmd ) , "<STR_LIT>" ) <EOL> class TestCombineStderr ( FabricTest ) : <EOL> @ server ( ) <EOL> def test_local_none_global_true ( self ) : <EOL> """<STR_LIT>""" <EOL> output . everything = False <EOL> r = run ( "<STR_LIT>" ) <EOL> eq_ ( "<STR_LIT>" , r . stdout ) <EOL> eq_ ( r . stderr , "<STR_LIT>" ) <EOL> @ server ( ) <EOL> def test_local_none_global_false ( self ) : <EOL> """<STR_LIT>""" <EOL> output . everything = False <EOL> env . combine_stderr = False <EOL> r = run ( "<STR_LIT>" ) <EOL> eq_ ( "<STR_LIT>" , r . stdout ) <EOL> eq_ ( "<STR_LIT>" , r . stderr ) <EOL> @ server ( ) <EOL> def test_local_true_global_false ( self ) : <EOL> """<STR_LIT>""" <EOL> output . everything = False <EOL> env . combine_stderr = False <EOL> r = run ( "<STR_LIT>" , combine_stderr = True ) <EOL> eq_ ( "<STR_LIT>" , r . stdout ) <EOL> eq_ ( r . stderr , "<STR_LIT>" ) <EOL> @ server ( ) <EOL> def test_local_false_global_true ( self ) : <EOL> """<STR_LIT>""" <EOL> output . everything = False <EOL> env . combine_stderr = True <EOL> r = run ( "<STR_LIT>" , combine_stderr = False ) <EOL> eq_ ( "<STR_LIT>" , r . stdout ) <EOL> eq_ ( "<STR_LIT>" , r . stderr ) <EOL> class TestQuietAndWarnKwargs ( FabricTest ) : <EOL> @ server ( responses = { '<STR_LIT>' : [ "<STR_LIT>" , "<STR_LIT>" , <NUM_LIT:1> ] } ) <EOL> def test_quiet_implies_warn_only ( self ) : <EOL> eq_ ( run ( "<STR_LIT>" , quiet = True ) . failed , True ) <EOL> @ server ( ) <EOL> @ mock_streams ( '<STR_LIT>' ) <EOL> def test_quiet_implies_hide_everything ( self ) : <EOL> run ( "<STR_LIT>" , quiet = True ) <EOL> eq_ ( sys . stdout . getvalue ( ) , "<STR_LIT>" ) <EOL> eq_ ( sys . stderr . getvalue ( ) , "<STR_LIT>" ) <EOL> @ server ( responses = { '<STR_LIT>' : [ "<STR_LIT>" , "<STR_LIT>" , <NUM_LIT:1> ] } ) <EOL> @ mock_streams ( '<STR_LIT>' ) <EOL> def test_warn_only_is_same_as_settings_warn_only ( self ) : <EOL> eq_ ( run ( "<STR_LIT>" , warn_only = True ) . failed , True ) <EOL> @ server ( ) <EOL> @ mock_streams ( '<STR_LIT>' ) <EOL> def test_warn_only_does_not_imply_hide_everything ( self ) : <EOL> run ( "<STR_LIT>" , warn_only = True ) <EOL> assert sys . stdout . getvalue ( ) != "<STR_LIT>" <EOL> class TestMultipleOKReturnCodes ( FabricTest ) : <EOL> @ server ( responses = { '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT:1> ] } ) <EOL> def test_expand_to_include_1 ( self ) : <EOL> with settings ( quiet ( ) , ok_ret_codes = [ <NUM_LIT:0> , <NUM_LIT:1> ] ) : <EOL> eq_ ( run ( "<STR_LIT>" ) . succeeded , True ) <EOL> slow_server = server ( responses = { '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT:3> ] } ) <EOL> slow = lambda x : slow_server ( raises ( CommandTimeout ) ( x ) ) <EOL> class TestRun ( FabricTest ) : <EOL> """<STR_LIT>""" <EOL> @ slow <EOL> def test_command_timeout_via_env_var ( self ) : <EOL> env . command_timeout = <NUM_LIT:2> <EOL> with hide ( '<STR_LIT>' ) : <EOL> run ( "<STR_LIT>" ) <EOL> @ slow <EOL> def test_command_timeout_via_kwarg ( self ) : <EOL> with hide ( '<STR_LIT>' ) : <EOL> run ( "<STR_LIT>" , timeout = <NUM_LIT:2> ) <EOL> @ slow <EOL> def test_command_timeout_via_env_var_in_sudo ( self ) : <EOL> env . command_timeout = <NUM_LIT:2> <EOL> with hide ( '<STR_LIT>' ) : <EOL> sudo ( "<STR_LIT>" ) <EOL> @ slow <EOL> def test_command_timeout_via_kwarg_of_sudo ( self ) : <EOL> with hide ( '<STR_LIT>' ) : <EOL> sudo ( "<STR_LIT>" , timeout = <NUM_LIT:2> ) <EOL> class TestFileTransfers ( FabricTest ) : <EOL> @ server ( files = { '<STR_LIT>' : '<STR_LIT>' } , home = '<STR_LIT>' ) <EOL> def test_get_relative_remote_dir_uses_home ( self ) : <EOL> """<STR_LIT>""" <EOL> with hide ( '<STR_LIT>' ) : <EOL> eq_ ( get ( '<STR_LIT>' , self . path ( ) ) , [ self . path ( '<STR_LIT>' ) ] ) <EOL> @ server ( files = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> def test_get_with_format_chars_on_server ( self ) : <EOL> """<STR_LIT>""" <EOL> remote = '<STR_LIT:*>' <EOL> with hide ( '<STR_LIT>' ) : <EOL> get ( remote , self . path ( ) ) <EOL> @ server ( ) <EOL> def test_get_single_file ( self ) : <EOL> """<STR_LIT>""" <EOL> remote = '<STR_LIT>' <EOL> local = self . path ( remote ) <EOL> with hide ( '<STR_LIT>' ) : <EOL> get ( remote , local ) <EOL> eq_contents ( local , FILES [ remote ] ) <EOL> @ server ( files = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> def test_get_file_from_relative_path_with_spaces ( self ) : <EOL> """<STR_LIT>""" <EOL> with hide ( '<STR_LIT>' ) : <EOL> with cd ( '<STR_LIT>' ) : <EOL> eq_ ( get ( '<STR_LIT:file>' , self . path ( ) ) , [ self . path ( '<STR_LIT:file>' ) ] ) <EOL> @ server ( ) <EOL> def test_get_sibling_globs ( self ) : <EOL> """<STR_LIT>""" <EOL> remotes = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> with hide ( '<STR_LIT>' ) : <EOL> get ( '<STR_LIT>' , self . tmpdir ) <EOL> for remote in remotes : <EOL> eq_contents ( self . path ( remote ) , FILES [ remote ] ) <EOL> @ server ( ) <EOL> def test_get_single_file_in_folder ( self ) : <EOL> """<STR_LIT>""" <EOL> remote = '<STR_LIT>' <EOL> with hide ( '<STR_LIT>' ) : <EOL> get ( '<STR_LIT>' , self . tmpdir ) <EOL> eq_contents ( self . path ( remote ) , FILES [ remote ] ) <EOL> @ server ( ) <EOL> def test_get_tree ( self ) : <EOL> """<STR_LIT>""" <EOL> with hide ( '<STR_LIT>' ) : <EOL> get ( '<STR_LIT>' , self . tmpdir ) <EOL> leaves = filter ( lambda x : x [ <NUM_LIT:0> ] . startswith ( '<STR_LIT>' ) , FILES . items ( ) ) <EOL> for path , contents in leaves : <EOL> eq_contents ( self . path ( path [ <NUM_LIT:1> : ] ) , contents ) <EOL> @ server ( ) <EOL> def test_get_tree_with_implicit_local_path ( self ) : <EOL> """<STR_LIT>""" <EOL> dirname = env . host_string . replace ( '<STR_LIT::>' , '<STR_LIT:->' ) <EOL> try : <EOL> with hide ( '<STR_LIT>' ) : <EOL> get ( '<STR_LIT>' ) <EOL> leaves = filter ( lambda x : x [ <NUM_LIT:0> ] . startswith ( '<STR_LIT>' ) , FILES . items ( ) ) <EOL> for path , contents in leaves : <EOL> path = os . path . join ( dirname , path [ <NUM_LIT:1> : ] ) <EOL> eq_contents ( path , contents ) <EOL> os . remove ( path ) <EOL> finally : <EOL> if os . path . exists ( dirname ) : <EOL> shutil . rmtree ( dirname ) <EOL> @ server ( ) <EOL> def test_get_absolute_path_should_save_relative ( self ) : <EOL> """<STR_LIT>""" <EOL> lpath = self . path ( ) <EOL> ltarget = os . path . join ( lpath , "<STR_LIT>" ) <EOL> with hide ( '<STR_LIT>' ) : <EOL> get ( '<STR_LIT>' , ltarget ) <EOL> assert self . exists_locally ( os . path . join ( lpath , '<STR_LIT>' ) ) <EOL> assert not self . exists_locally ( os . path . join ( lpath , '<STR_LIT>' ) ) <EOL> @ server ( ) <EOL> def test_path_formatstr_nonrecursively_is_just_filename ( self ) : <EOL> """<STR_LIT>""" <EOL> lpath = self . path ( ) <EOL> ltarget = os . path . join ( lpath , "<STR_LIT>" ) <EOL> with hide ( '<STR_LIT>' ) : <EOL> get ( '<STR_LIT>' , ltarget ) <EOL> assert self . exists_locally ( os . path . join ( lpath , '<STR_LIT>' ) ) <EOL> @ server ( ) <EOL> @ mock_streams ( '<STR_LIT>' ) <EOL> def _invalid_file_obj_situations ( self , remote_path ) : <EOL> with settings ( hide ( '<STR_LIT>' ) , warn_only = True ) : <EOL> get ( remote_path , StringIO ( ) ) <EOL> assert_contains ( '<STR_LIT>' , sys . stderr . getvalue ( ) ) <EOL> def test_glob_and_file_object_invalid ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _invalid_file_obj_situations ( '<STR_LIT>' ) <EOL> def test_directory_and_file_object_invalid ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _invalid_file_obj_situations ( '<STR_LIT>' ) <EOL> @ server ( ) <EOL> def test_get_single_file_absolutely ( self ) : <EOL> """<STR_LIT>""" <EOL> target = '<STR_LIT>' <EOL> with hide ( '<STR_LIT>' ) : <EOL> get ( target , self . tmpdir ) <EOL> eq_contents ( self . path ( os . path . basename ( target ) ) , FILES [ target ] ) <EOL> @ server ( ) <EOL> def test_get_file_with_nonexistent_target ( self ) : <EOL> """<STR_LIT>""" <EOL> local = self . path ( '<STR_LIT>' ) <EOL> target = '<STR_LIT>' <EOL> with hide ( '<STR_LIT>' ) : <EOL> get ( target , local ) <EOL> eq_contents ( local , FILES [ target ] ) <EOL> @ server ( ) <EOL> @ mock_streams ( '<STR_LIT>' ) <EOL> def test_get_file_with_existing_file_target ( self ) : <EOL> """<STR_LIT>""" <EOL> local = self . path ( '<STR_LIT>' ) <EOL> target = '<STR_LIT>' <EOL> with open ( local , '<STR_LIT:w>' ) as fd : <EOL> fd . write ( "<STR_LIT:foo>" ) <EOL> with hide ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> get ( target , local ) <EOL> assert "<STR_LIT>" % local in sys . stderr . getvalue ( ) <EOL> eq_contents ( local , FILES [ target ] ) <EOL> @ server ( ) <EOL> def test_get_file_to_directory ( self ) : <EOL> """<STR_LIT>""" <EOL> target = '<STR_LIT>' <EOL> with hide ( '<STR_LIT>' ) : <EOL> get ( target , self . tmpdir ) <EOL> eq_contents ( self . path ( target ) , FILES [ target ] ) <EOL> @ server ( port = <NUM_LIT> ) <EOL> @ server ( port = <NUM_LIT> ) <EOL> def test_get_from_multiple_servers ( self ) : <EOL> ports = [ <NUM_LIT> , <NUM_LIT> ] <EOL> hosts = map ( lambda x : '<STR_LIT>' % x , ports ) <EOL> with settings ( all_hosts = hosts ) : <EOL> for port in ports : <EOL> with settings ( <EOL> hide ( '<STR_LIT>' ) , host_string = '<STR_LIT>' % port <EOL> ) : <EOL> tmp = self . path ( '<STR_LIT>' ) <EOL> local_path = os . path . join ( tmp , "<STR_LIT>" , "<STR_LIT>" ) <EOL> path = '<STR_LIT>' <EOL> get ( path , local_path ) <EOL> assert self . exists_locally ( os . path . join ( <EOL> tmp , "<STR_LIT>" % port , path <EOL> ) ) <EOL> get ( '<STR_LIT>' , local_path ) <EOL> assert self . exists_locally ( os . path . join ( <EOL> tmp , "<STR_LIT>" % port , '<STR_LIT>' <EOL> ) ) <EOL> @ server ( ) <EOL> def test_get_from_empty_directory_uses_cwd ( self ) : <EOL> """<STR_LIT>""" <EOL> with hide ( '<STR_LIT>' ) : <EOL> get ( '<STR_LIT>' , self . tmpdir ) <EOL> for x in "<STR_LIT>" . split ( ) : <EOL> assert os . path . exists ( os . path . join ( self . tmpdir , x ) ) <EOL> @ server ( ) <EOL> def _get_to_cwd ( self , arg ) : <EOL> path = '<STR_LIT>' <EOL> with hide ( '<STR_LIT>' ) : <EOL> get ( path , arg ) <EOL> host_dir = os . path . join ( <EOL> os . getcwd ( ) , <EOL> env . host_string . replace ( '<STR_LIT::>' , '<STR_LIT:->' ) , <EOL> ) <EOL> target = os . path . join ( host_dir , path ) <EOL> try : <EOL> assert os . path . exists ( target ) <EOL> finally : <EOL> shutil . rmtree ( host_dir ) <EOL> def test_get_to_empty_string_uses_default_format_string ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _get_to_cwd ( '<STR_LIT>' ) <EOL> def test_get_to_None_uses_default_format_string ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _get_to_cwd ( None ) <EOL> @ server ( ) <EOL> def test_get_should_accept_file_like_objects ( self ) : <EOL> """<STR_LIT>""" <EOL> fake_file = StringIO ( ) <EOL> target = '<STR_LIT>' <EOL> with hide ( '<STR_LIT>' ) : <EOL> get ( target , fake_file ) <EOL> eq_ ( fake_file . getvalue ( ) , FILES [ target ] ) <EOL> @ server ( ) <EOL> def test_get_interpolation_without_host ( self ) : <EOL> """<STR_LIT>""" <EOL> with hide ( '<STR_LIT>' ) : <EOL> tmp = self . path ( '<STR_LIT>' ) <EOL> local_path = tmp + "<STR_LIT>" <EOL> get ( '<STR_LIT>' , local_path ) <EOL> assert self . exists_locally ( tmp + "<STR_LIT>" ) <EOL> local_path = tmp + "<STR_LIT>" <EOL> get ( '<STR_LIT>' , local_path ) <EOL> assert self . exists_locally ( tmp + "<STR_LIT>" ) <EOL> @ server ( ) <EOL> def test_get_returns_list_of_local_paths ( self ) : <EOL> """<STR_LIT>""" <EOL> d = self . path ( ) <EOL> with hide ( '<STR_LIT>' ) : <EOL> retval = get ( '<STR_LIT>' , d ) <EOL> files = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> eq_ ( map ( lambda x : os . path . join ( d , '<STR_LIT>' , x ) , files ) , retval ) <EOL> @ server ( ) <EOL> def test_get_returns_none_for_stringio ( self ) : <EOL> """<STR_LIT>""" <EOL> with hide ( '<STR_LIT>' ) : <EOL> eq_ ( [ ] , get ( '<STR_LIT>' , StringIO ( ) ) ) <EOL> @ server ( ) <EOL> def test_get_return_value_failed_attribute ( self ) : <EOL> """<STR_LIT>""" <EOL> with settings ( hide ( '<STR_LIT>' ) , warn_only = True ) : <EOL> retval = get ( '<STR_LIT>' , self . path ( ) ) <EOL> eq_ ( [ '<STR_LIT>' ] , retval . failed ) <EOL> assert not retval . succeeded <EOL> @ server ( ) <EOL> def test_get_should_not_use_windows_slashes_in_remote_paths ( self ) : <EOL> """<STR_LIT>""" <EOL> with hide ( '<STR_LIT>' ) : <EOL> path = "<STR_LIT>" <EOL> sftp = SFTP ( env . host_string ) <EOL> eq_ ( sftp . glob ( path ) , [ path ] ) <EOL> @ server ( ) <EOL> @ with_fakes <EOL> def test_get_use_sudo ( self ) : <EOL> """<STR_LIT>""" <EOL> name = "<STR_LIT>" <EOL> fake_run = Fake ( '<STR_LIT>' , callable = True , expect_call = True ) . with_matching_args ( <EOL> '<STR_LIT>' % name , True , True , None , <EOL> ) . next_call ( ) . with_matching_args ( <EOL> '<STR_LIT>' % name , True , True , None , <EOL> ) . next_call ( ) . with_matching_args ( <EOL> '<STR_LIT>' % name , True , True , None , <EOL> ) . next_call ( ) . with_matching_args ( <EOL> '<STR_LIT>' % name , True , True , None , <EOL> ) <EOL> fake_get = Fake ( '<STR_LIT>' , callable = True , expect_call = True ) . with_args ( <EOL> name , fudge_arg . any_value ( ) ) <EOL> with hide ( '<STR_LIT>' ) : <EOL> with patched_context ( '<STR_LIT>' , '<STR_LIT>' , fake_run ) : <EOL> with patched_context ( SFTPClient , '<STR_LIT>' , fake_get ) : <EOL> retval = get ( '<STR_LIT>' , self . path ( ) , use_sudo = True ) <EOL> assert retval [ <NUM_LIT:0> ] . endswith ( '<STR_LIT>' ) <EOL> @ server ( ) <EOL> @ with_fakes <EOL> def test_get_use_sudo_temp_dir ( self ) : <EOL> """<STR_LIT>""" <EOL> name = "<STR_LIT>" <EOL> fake_run = Fake ( '<STR_LIT>' , callable = True , expect_call = True ) . with_matching_args ( <EOL> '<STR_LIT>' % name , True , True , None , <EOL> ) . next_call ( ) . with_matching_args ( <EOL> '<STR_LIT>' % name , True , True , None , <EOL> ) . next_call ( ) . with_matching_args ( <EOL> '<STR_LIT>' % name , True , True , None , <EOL> ) . next_call ( ) . with_matching_args ( <EOL> '<STR_LIT>' % name , True , True , None , <EOL> ) <EOL> fake_get = Fake ( '<STR_LIT>' , callable = True , expect_call = True ) . with_args ( <EOL> '<STR_LIT>' % name , fudge_arg . any_value ( ) ) <EOL> with hide ( '<STR_LIT>' ) : <EOL> with patched_context ( '<STR_LIT>' , '<STR_LIT>' , fake_run ) : <EOL> with patched_context ( SFTPClient , '<STR_LIT>' , fake_get ) : <EOL> retval = get ( '<STR_LIT>' , self . path ( ) , use_sudo = True , temp_dir = "<STR_LIT>" ) <EOL> assert retval [ <NUM_LIT:0> ] . endswith ( '<STR_LIT>' ) <EOL> @ server ( ) <EOL> def test_put_file_to_existing_directory ( self ) : <EOL> """<STR_LIT>""" <EOL> text = "<STR_LIT>" <EOL> local = self . mkfile ( '<STR_LIT>' , text ) <EOL> local2 = self . path ( '<STR_LIT>' ) <EOL> with hide ( '<STR_LIT>' ) : <EOL> put ( local , '<STR_LIT:/>' ) <EOL> get ( '<STR_LIT>' , local2 ) <EOL> eq_contents ( local2 , text ) <EOL> @ server ( ) <EOL> def test_put_to_empty_directory_uses_cwd ( self ) : <EOL> """<STR_LIT>""" <EOL> text = "<STR_LIT>" <EOL> local = self . path ( '<STR_LIT>' ) <EOL> local2 = self . path ( '<STR_LIT>' ) <EOL> with open ( local , '<STR_LIT:w>' ) as fd : <EOL> fd . write ( text ) <EOL> with hide ( '<STR_LIT>' ) : <EOL> put ( local ) <EOL> get ( '<STR_LIT>' , local2 ) <EOL> eq_contents ( local2 , text ) <EOL> @ server ( ) <EOL> def test_put_from_empty_directory_uses_cwd ( self ) : <EOL> """<STR_LIT>""" <EOL> text = '<STR_LIT>' <EOL> old_cwd = os . getcwd ( ) <EOL> os . chdir ( self . tmpdir ) <EOL> with open ( '<STR_LIT>' , '<STR_LIT:w>' ) as fd : <EOL> fd . write ( text ) <EOL> with hide ( '<STR_LIT>' ) : <EOL> put ( '<STR_LIT>' , '<STR_LIT:/>' ) <EOL> remote = os . path . join ( os . path . basename ( self . tmpdir ) , '<STR_LIT>' ) <EOL> get ( remote , '<STR_LIT>' ) <EOL> eq_contents ( '<STR_LIT>' , text ) <EOL> os . chdir ( old_cwd ) <EOL> @ server ( ) <EOL> def test_put_should_accept_file_like_objects ( self ) : <EOL> """<STR_LIT>""" <EOL> local = self . path ( '<STR_LIT>' ) <EOL> fake_file = StringIO ( ) <EOL> fake_file . write ( "<STR_LIT>" ) <EOL> pointer = fake_file . tell ( ) <EOL> target = '<STR_LIT>' <EOL> with hide ( '<STR_LIT>' ) : <EOL> put ( fake_file , target ) <EOL> get ( target , local ) <EOL> eq_contents ( local , fake_file . getvalue ( ) ) <EOL> eq_ ( pointer , fake_file . tell ( ) ) <EOL> @ server ( ) <EOL> @ raises ( ValueError ) <EOL> def test_put_should_raise_exception_for_nonexistent_local_path ( self ) : <EOL> """<STR_LIT>""" <EOL> put ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> @ server ( ) <EOL> def test_put_returns_list_of_remote_paths ( self ) : <EOL> """<STR_LIT>""" <EOL> p = '<STR_LIT>' <EOL> f = self . path ( p ) <EOL> with open ( f , '<STR_LIT:w>' ) as fd : <EOL> fd . write ( "<STR_LIT>" ) <EOL> with hide ( '<STR_LIT>' ) : <EOL> retval = put ( f , p ) <EOL> eq_ ( retval , [ p ] ) <EOL> @ server ( ) <EOL> def test_put_returns_list_of_remote_paths_with_stringio ( self ) : <EOL> """<STR_LIT>""" <EOL> f = '<STR_LIT>' <EOL> with hide ( '<STR_LIT>' ) : <EOL> eq_ ( put ( StringIO ( '<STR_LIT>' ) , f ) , [ f ] ) <EOL> @ server ( ) <EOL> def test_put_return_value_failed_attribute ( self ) : <EOL> """<STR_LIT>""" <EOL> with settings ( hide ( '<STR_LIT>' ) , warn_only = True ) : <EOL> f = StringIO ( '<STR_LIT>' ) <EOL> retval = put ( f , '<STR_LIT>' ) <EOL> eq_ ( [ "<STR_LIT>" ] , retval . failed ) <EOL> assert not retval . succeeded <EOL> @ server ( ) <EOL> def test_put_sends_all_files_with_glob ( self ) : <EOL> """<STR_LIT>""" <EOL> paths = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> glob = '<STR_LIT>' <EOL> remote_directory = '<STR_LIT:/>' <EOL> for path in paths : <EOL> self . mkfile ( path , '<STR_LIT>' ) <EOL> with hide ( '<STR_LIT>' ) : <EOL> retval = put ( self . path ( glob ) , remote_directory ) <EOL> eq_ ( sorted ( retval ) , sorted ( [ remote_directory + path for path in paths ] ) ) <EOL> @ server ( ) <EOL> def test_put_sends_correct_file_with_globbing_off ( self ) : <EOL> """<STR_LIT>""" <EOL> text = "<STR_LIT>" <EOL> local = self . mkfile ( '<STR_LIT>' , text ) <EOL> local2 = self . path ( '<STR_LIT>' ) <EOL> with hide ( '<STR_LIT>' ) : <EOL> put ( local , '<STR_LIT:/>' , use_glob = False ) <EOL> get ( '<STR_LIT>' , local2 ) <EOL> eq_contents ( local2 , text ) <EOL> @ server ( ) <EOL> @ with_fakes <EOL> def test_put_use_sudo ( self ) : <EOL> """<STR_LIT>""" <EOL> fake_run = Fake ( '<STR_LIT>' , callable = True , expect_call = True ) . with_matching_args ( <EOL> '<STR_LIT>' , True , True , None , <EOL> ) <EOL> fake_put = Fake ( '<STR_LIT>' , callable = True , expect_call = True ) . with_args ( fudge_arg . any_value ( ) , <EOL> '<STR_LIT>' ) <EOL> local_path = self . mkfile ( '<STR_LIT>' , "<STR_LIT>" ) <EOL> with hide ( '<STR_LIT>' ) : <EOL> with patched_context ( '<STR_LIT>' , '<STR_LIT>' , fake_run ) : <EOL> with patched_context ( SFTPClient , '<STR_LIT>' , fake_put ) : <EOL> retval = put ( local_path , "<STR_LIT:/>" , use_sudo = True ) <EOL> assert retval [ <NUM_LIT:0> ] . endswith ( '<STR_LIT>' ) <EOL> @ server ( ) <EOL> @ with_fakes <EOL> def test_put_use_sudo_temp_dir ( self ) : <EOL> """<STR_LIT>""" <EOL> fake_run = Fake ( '<STR_LIT>' , callable = True , expect_call = True ) . with_matching_args ( <EOL> '<STR_LIT>' , True , True , None , <EOL> ) <EOL> fake_put = Fake ( '<STR_LIT>' , callable = True , expect_call = True ) . with_args ( fudge_arg . any_value ( ) , <EOL> '<STR_LIT>' ) <EOL> local_path = self . mkfile ( '<STR_LIT>' , "<STR_LIT>" ) <EOL> with hide ( '<STR_LIT>' ) : <EOL> with patched_context ( '<STR_LIT>' , '<STR_LIT>' , fake_run ) : <EOL> with patched_context ( SFTPClient , '<STR_LIT>' , fake_put ) : <EOL> retval = put ( local_path , "<STR_LIT:/>" , use_sudo = True , temp_dir = '<STR_LIT>' ) <EOL> assert retval [ <NUM_LIT:0> ] . endswith ( '<STR_LIT>' ) <EOL> @ server ( ) <EOL> def test_cd_should_apply_to_put ( self ) : <EOL> """<STR_LIT>""" <EOL> f = '<STR_LIT>' <EOL> d = '<STR_LIT>' <EOL> local = self . path ( f ) <EOL> with open ( local , '<STR_LIT:w>' ) as fd : <EOL> fd . write ( '<STR_LIT:test>' ) <EOL> with nested ( cd ( d ) , hide ( '<STR_LIT>' ) ) : <EOL> put ( local , f ) <EOL> assert self . exists_remotely ( '<STR_LIT>' % ( d , f ) ) <EOL> @ server ( files = { '<STR_LIT>' : '<STR_LIT:test>' } ) <EOL> def test_cd_should_apply_to_get ( self ) : <EOL> """<STR_LIT>""" <EOL> local = self . path ( '<STR_LIT>' ) <EOL> with nested ( cd ( '<STR_LIT>' ) , hide ( '<STR_LIT>' ) ) : <EOL> get ( '<STR_LIT>' , local ) <EOL> assert os . path . exists ( local ) <EOL> @ server ( ) <EOL> def test_cd_should_not_apply_to_absolute_put ( self ) : <EOL> """<STR_LIT>""" <EOL> local = self . path ( '<STR_LIT>' ) <EOL> with open ( local , '<STR_LIT:w>' ) as fd : <EOL> fd . write ( '<STR_LIT:test>' ) <EOL> with nested ( cd ( '<STR_LIT>' ) , hide ( '<STR_LIT>' ) ) : <EOL> put ( local , '<STR_LIT>' ) <EOL> assert not self . exists_remotely ( '<STR_LIT>' ) <EOL> assert self . exists_remotely ( '<STR_LIT>' ) <EOL> @ server ( files = { '<STR_LIT>' : '<STR_LIT:test>' } ) <EOL> def test_cd_should_not_apply_to_absolute_get ( self ) : <EOL> """<STR_LIT>""" <EOL> local = self . path ( '<STR_LIT>' ) <EOL> with nested ( cd ( '<STR_LIT>' ) , hide ( '<STR_LIT>' ) ) : <EOL> get ( '<STR_LIT>' , local ) <EOL> assert os . path . exists ( local ) <EOL> @ server ( ) <EOL> def test_lcd_should_apply_to_put ( self ) : <EOL> """<STR_LIT>""" <EOL> f = '<STR_LIT>' <EOL> d = '<STR_LIT>' <EOL> local = self . path ( d , f ) <EOL> os . makedirs ( os . path . dirname ( local ) ) <EOL> with open ( local , '<STR_LIT:w>' ) as fd : <EOL> fd . write ( "<STR_LIT>" ) <EOL> with nested ( lcd ( self . path ( d ) ) , hide ( '<STR_LIT>' ) ) : <EOL> put ( f , '<STR_LIT:/>' ) <EOL> assert self . exists_remotely ( '<STR_LIT>' % f ) <EOL> @ server ( ) <EOL> def test_lcd_should_apply_to_get ( self ) : <EOL> """<STR_LIT>""" <EOL> d = self . path ( '<STR_LIT>' ) <EOL> f = '<STR_LIT>' <EOL> with nested ( lcd ( d ) , hide ( '<STR_LIT>' ) ) : <EOL> get ( f , f ) <EOL> assert self . exists_locally ( os . path . join ( d , f ) ) <EOL> @ server ( ) <EOL> @ mock_streams ( '<STR_LIT>' ) <EOL> def test_stringio_without_name ( self ) : <EOL> file_obj = StringIO ( u'<STR_LIT>' ) <EOL> put ( file_obj , '<STR_LIT:/>' ) <EOL> assert re . search ( '<STR_LIT>' , sys . stdout . getvalue ( ) ) <EOL> @ server ( ) <EOL> @ mock_streams ( '<STR_LIT>' ) <EOL> def test_stringio_with_name ( self ) : <EOL> """<STR_LIT>""" <EOL> file_obj = StringIO ( u'<STR_LIT>' ) <EOL> file_obj . name = '<STR_LIT>' <EOL> put ( file_obj , '<STR_LIT:/>' ) <EOL> assert re . search ( file_obj . name , sys . stdout . getvalue ( ) ) <EOL> def test_local_output_and_capture ( ) : <EOL> for capture in ( True , False ) : <EOL> for stdout in ( True , False ) : <EOL> for stderr in ( True , False ) : <EOL> hides , shows = [ '<STR_LIT>' ] , [ ] <EOL> if stdout : <EOL> hides . append ( '<STR_LIT>' ) <EOL> else : <EOL> shows . append ( '<STR_LIT>' ) <EOL> if stderr : <EOL> hides . append ( '<STR_LIT>' ) <EOL> else : <EOL> shows . append ( '<STR_LIT>' ) <EOL> with nested ( hide ( * hides ) , show ( * shows ) ) : <EOL> d = "<STR_LIT>" % ( <EOL> capture , stdout , stderr <EOL> ) <EOL> local . description = d <EOL> yield local , "<STR_LIT>" , capture <EOL> del local . description <EOL> class TestRunSudoReturnValues ( FabricTest ) : <EOL> @ server ( ) <EOL> def test_returns_command_given ( self ) : <EOL> """<STR_LIT>""" <EOL> with hide ( '<STR_LIT>' ) : <EOL> eq_ ( run ( "<STR_LIT>" ) . command , "<STR_LIT>" ) <EOL> @ server ( ) <EOL> def test_returns_fully_wrapped_command ( self ) : <EOL> """<STR_LIT>""" <EOL> with settings ( hide ( '<STR_LIT>' ) , warn_only = True , use_shell = True ) : <EOL> ok_ ( env . shell in run ( "<STR_LIT>" ) . real_command ) </s>
