<s> """<STR_LIT>""" <EOL> from mock import patch <EOL> import os <EOL> from os . path import dirname , join as opj <EOL> import sys <EOL> import logging <EOL> import shlex <EOL> from . utils import ok_ , eq_ , assert_is , assert_equal , assert_false , assert_true , assert_greater , assert_raises , assert_in , SkipTest <EOL> from . . cmd import Runner , link_file_load <EOL> from . . support . exceptions import CommandError <EOL> from . . support . protocol import DryRunProtocol <EOL> from . utils import with_tempfile , assert_cwd_unchanged , ignore_nose_capturing_stdout , swallow_outputs , swallow_logs , on_linux , on_osx , on_windows , with_testrepos <EOL> from . utils import lgr <EOL> from . utils import local_testrepo_flavors <EOL> @ ignore_nose_capturing_stdout <EOL> @ assert_cwd_unchanged <EOL> @ with_tempfile <EOL> def test_runner_dry ( tempfile ) : <EOL> dry = DryRunProtocol ( ) <EOL> runner = Runner ( protocol = dry ) <EOL> cmd = '<STR_LIT>' % tempfile <EOL> with swallow_logs ( new_level = logging . DEBUG ) as cml : <EOL> ret = runner . run ( cmd ) <EOL> assert_equal ( cml . out . rstrip ( ) , "<STR_LIT>" % cmd ) <EOL> assert_equal ( ( "<STR_LIT>" , "<STR_LIT>" ) , ret , <EOL> "<STR_LIT>" % ( cmd , ret ) ) <EOL> assert_equal ( shlex . split ( cmd , posix = not on_windows ) , dry [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> assert_false ( os . path . exists ( tempfile ) ) <EOL> output = runner . call ( os . path . join , '<STR_LIT:foo>' , '<STR_LIT:bar>' ) <EOL> assert_is ( None , output , "<STR_LIT>" <EOL> "<STR_LIT>" % output ) <EOL> assert_in ( '<STR_LIT>' , dry [ <NUM_LIT:1> ] [ '<STR_LIT>' ] [ <NUM_LIT:0> ] ) <EOL> assert_equal ( "<STR_LIT>" , dry [ <NUM_LIT:1> ] [ '<STR_LIT>' ] [ <NUM_LIT:1> ] ) <EOL> @ ignore_nose_capturing_stdout <EOL> @ assert_cwd_unchanged <EOL> @ with_tempfile <EOL> def test_runner ( tempfile ) : <EOL> runner = Runner ( ) <EOL> cmd = '<STR_LIT>' % tempfile <EOL> ret = runner . run ( cmd ) <EOL> assert_true ( os . path . exists ( tempfile ) , <EOL> "<STR_LIT>" % <EOL> ( cmd , tempfile ) ) <EOL> output = runner . call ( os . path . join , '<STR_LIT:foo>' , '<STR_LIT:bar>' ) <EOL> assert_equal ( os . path . join ( '<STR_LIT:foo>' , '<STR_LIT:bar>' ) , output , <EOL> "<STR_LIT>" % output ) <EOL> @ ignore_nose_capturing_stdout <EOL> def test_runner_instance_callable_dry ( ) : <EOL> cmd_ = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:string>' ] <EOL> for cmd in [ cmd_ , '<STR_LIT:U+0020>' . join ( cmd_ ) ] : <EOL> dry = DryRunProtocol ( ) <EOL> runner = Runner ( protocol = dry ) <EOL> ret = runner ( cmd ) <EOL> eq_ ( ret , ( "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> assert_equal ( cmd_ , dry [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % ( cmd , dry ) ) <EOL> ret = runner ( os . path . join , '<STR_LIT:foo>' , '<STR_LIT:bar>' ) <EOL> eq_ ( ret , None ) <EOL> assert_in ( '<STR_LIT>' , dry [ <NUM_LIT:1> ] [ '<STR_LIT>' ] [ <NUM_LIT:0> ] , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % dry ) <EOL> assert_equal ( "<STR_LIT>" , dry [ <NUM_LIT:1> ] [ '<STR_LIT>' ] [ <NUM_LIT:1> ] , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % dry ) <EOL> @ ignore_nose_capturing_stdout <EOL> def test_runner_instance_callable_wet ( ) : <EOL> runner = Runner ( ) <EOL> cmd = [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] <EOL> out = runner ( cmd ) <EOL> eq_ ( out [ <NUM_LIT:0> ] . rstrip ( ) , ( '<STR_LIT>' ) ) <EOL> eq_ ( out [ <NUM_LIT:1> ] , '<STR_LIT>' ) <EOL> ret = runner ( os . path . join , '<STR_LIT:foo>' , '<STR_LIT:bar>' ) <EOL> eq_ ( ret , os . path . join ( '<STR_LIT:foo>' , '<STR_LIT:bar>' ) ) <EOL> @ ignore_nose_capturing_stdout <EOL> def test_runner_log_stderr ( ) : <EOL> runner = Runner ( ) <EOL> cmd = '<STR_LIT>' <EOL> ret = runner . run ( cmd , log_stderr = True , expect_stderr = True ) <EOL> cmd = '<STR_LIT>' <EOL> with swallow_outputs ( ) as cmo : <EOL> with swallow_logs ( new_level = logging . INFO ) as cml : <EOL> ret = runner . run ( cmd , log_stderr = False ) <EOL> eq_ ( cmo . err . rstrip ( ) , "<STR_LIT>" ) <EOL> eq_ ( cml . out , "<STR_LIT>" ) <EOL> @ ignore_nose_capturing_stdout <EOL> def test_runner_log_stdout ( ) : <EOL> runner = Runner ( ) <EOL> cmd_ = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> for cmd in [ cmd_ , '<STR_LIT:U+0020>' . join ( cmd_ ) ] : <EOL> kw = { } <EOL> if on_windows and isinstance ( cmd , list ) : <EOL> kw [ '<STR_LIT>' ] = True <EOL> with swallow_logs ( logging . DEBUG ) as cm : <EOL> ret = runner . run ( cmd , log_stdout = True , ** kw ) <EOL> eq_ ( cm . lines [ <NUM_LIT:0> ] , "<STR_LIT>" % cmd ) <EOL> if not on_windows : <EOL> eq_ ( cm . lines [ <NUM_LIT:1> ] , "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> else : <EOL> ok_ ( "<STR_LIT>" in cm . lines [ <NUM_LIT:1> ] ) <EOL> cmd = '<STR_LIT>' <EOL> with swallow_outputs ( ) as cmo : <EOL> with swallow_logs ( new_level = logging . INFO ) as cml : <EOL> ret = runner . run ( cmd , log_stdout = False ) <EOL> eq_ ( cmo . out , "<STR_LIT>" ) <EOL> eq_ ( cml . out , "<STR_LIT>" ) <EOL> @ ignore_nose_capturing_stdout <EOL> def check_runner_heavy_output ( log_online ) : <EOL> runner = Runner ( ) <EOL> cmd = '<STR_LIT>' % ( sys . executable , opj ( dirname ( __file__ ) , "<STR_LIT>" ) ) <EOL> with swallow_outputs ( ) as cm , swallow_logs ( ) : <EOL> ret = runner . run ( cmd , log_stderr = False , log_stdout = False , <EOL> expect_stderr = True ) <EOL> eq_ ( cm . err , cm . out ) <EOL> eq_ ( cm . out [ : <NUM_LIT:10> ] , "<STR_LIT>" ) <EOL> eq_ ( cm . out [ - <NUM_LIT:15> : ] , "<STR_LIT>" ) <EOL> if lgr . getEffectiveLevel ( ) <= logging . DEBUG : <EOL> raise SkipTest ( "<STR_LIT>" ) <EOL> with swallow_logs ( ) : <EOL> ret = runner . run ( cmd , log_stderr = True , log_stdout = True , expect_stderr = True ) <EOL> return <EOL> if not log_online : <EOL> cmd = '<STR_LIT>' '<STR_LIT>' '<STR_LIT>' % sys . executable <EOL> with swallow_logs ( ) : <EOL> ret = runner . run ( cmd , log_stderr = True , log_stdout = True , <EOL> expect_stderr = True ) <EOL> def test_runner_heavy_output ( ) : <EOL> for log_online in [ True , False ] : <EOL> yield check_runner_heavy_output , log_online <EOL> @ with_tempfile <EOL> def test_link_file_load ( tempfile ) : <EOL> tempfile2 = tempfile + '<STR_LIT:_>' <EOL> with open ( tempfile , '<STR_LIT:w>' ) as f : <EOL> f . write ( "<STR_LIT>" ) <EOL> link_file_load ( tempfile , tempfile2 ) <EOL> ok_ ( os . path . exists ( tempfile2 ) ) <EOL> with open ( tempfile2 , '<STR_LIT:r>' ) as f : <EOL> assert_equal ( f . read ( ) , "<STR_LIT>" ) <EOL> def inode ( fname ) : <EOL> with open ( fname ) as fd : <EOL> return os . fstat ( fd . fileno ( ) ) . st_ino <EOL> def stats ( fname , times = True ) : <EOL> """<STR_LIT>""" <EOL> with open ( fname ) as fd : <EOL> st = os . fstat ( fd . fileno ( ) ) <EOL> stats = ( st . st_mode , st . st_uid , st . st_gid , st . st_size ) <EOL> if times : <EOL> return stats + ( st . st_atime , st . st_mtime ) <EOL> else : <EOL> return stats <EOL> if on_linux or on_osx : <EOL> assert_equal ( inode ( tempfile ) , inode ( tempfile2 ) ) <EOL> assert_equal ( stats ( tempfile ) , stats ( tempfile2 ) ) <EOL> def raise_AttributeError ( * args ) : <EOL> raise AttributeError ( "<STR_LIT>" ) <EOL> with patch ( '<STR_LIT>' , raise_AttributeError ) : <EOL> with swallow_logs ( logging . WARNING ) as cm : <EOL> link_file_load ( tempfile , tempfile2 ) <EOL> ok_ ( "<STR_LIT>" in cm . out ) <EOL> ok_ ( inode ( tempfile ) != inode ( tempfile2 ) ) <EOL> with open ( tempfile2 , '<STR_LIT:r>' ) as f : <EOL> assert_equal ( f . read ( ) , "<STR_LIT>" ) <EOL> assert_equal ( stats ( tempfile , times = False ) , stats ( tempfile2 , times = False ) ) <EOL> os . unlink ( tempfile2 ) <EOL> @ with_tempfile ( mkdir = True ) <EOL> def test_runner_failure ( dir_ ) : <EOL> from . . support . annexrepo import AnnexRepo <EOL> repo = AnnexRepo ( dir_ , create = True ) <EOL> runner = Runner ( ) <EOL> failing_cmd = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> with assert_raises ( CommandError ) as cme , swallow_logs ( ) as cml : <EOL> runner . run ( failing_cmd , cwd = dir_ ) <EOL> assert_in ( '<STR_LIT>' , cml . out ) <EOL> assert_equal ( <NUM_LIT:1> , cme . exception . code ) </s>
