<s> """<STR_LIT>""" <EOL> import __builtin__ <EOL> import errno <EOL> import os <EOL> import posix <EOL> import pwd <EOL> import shutil <EOL> import stat <EOL> import tempfile <EOL> import mox <EOL> from google . apputils import basetest <EOL> from google . apputils import file_util <EOL> import gflags as flags <EOL> FLAGS = flags . FLAGS <EOL> class FileUtilTest ( basetest . TestCase ) : <EOL> def testHomeDir ( self ) : <EOL> self . assertEqual ( file_util . HomeDir ( ) , pwd . getpwuid ( os . geteuid ( ) ) . pw_dir ) <EOL> self . assertEqual ( file_util . HomeDir ( <NUM_LIT:0> ) , pwd . getpwuid ( <NUM_LIT:0> ) . pw_dir ) <EOL> self . assertEqual ( file_util . HomeDir ( '<STR_LIT:root>' ) , pwd . getpwnam ( '<STR_LIT:root>' ) . pw_dir ) <EOL> class FileUtilTempdirTest ( basetest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . temp_dir = tempfile . mkdtemp ( ) <EOL> self . file_path = self . temp_dir + '<STR_LIT>' <EOL> self . sample_contents = '<STR_LIT>' <EOL> self . prev_umask = posix . umask ( <NUM_LIT:0> ) <EOL> def tearDown ( self ) : <EOL> shutil . rmtree ( self . temp_dir ) <EOL> posix . umask ( self . prev_umask ) <EOL> def testWriteOverwrite ( self ) : <EOL> file_util . Write ( self . file_path , '<STR_LIT>' ) <EOL> file_util . Write ( self . file_path , self . sample_contents ) <EOL> with open ( self . file_path ) as fp : <EOL> self . assertEquals ( fp . read ( ) , self . sample_contents ) <EOL> def testWriteExclusive ( self ) : <EOL> file_util . Write ( self . file_path , '<STR_LIT>' ) <EOL> self . assertRaises ( OSError , file_util . Write , self . file_path , <EOL> self . sample_contents , overwrite_existing = False ) <EOL> def testWriteMode ( self ) : <EOL> mode = <NUM_LIT:0> <NUM_LIT> <EOL> file_util . Write ( self . file_path , self . sample_contents , mode = mode ) <EOL> s = os . stat ( self . file_path ) <EOL> self . assertEqual ( stat . S_IMODE ( s . st_mode ) , mode ) <EOL> def testAtomicWriteSuccessful ( self ) : <EOL> file_util . AtomicWrite ( self . file_path , self . sample_contents ) <EOL> with open ( self . file_path ) as fp : <EOL> self . assertEquals ( fp . read ( ) , self . sample_contents ) <EOL> def testAtomicWriteMode ( self ) : <EOL> mode = <NUM_LIT:0> <NUM_LIT> <EOL> file_util . AtomicWrite ( self . file_path , self . sample_contents , mode = mode ) <EOL> s = os . stat ( self . file_path ) <EOL> self . assertEqual ( stat . S_IMODE ( s . st_mode ) , mode ) <EOL> class FileUtilMoxTestBase ( basetest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . mox = mox . Mox ( ) <EOL> self . sample_contents = '<STR_LIT>' <EOL> self . file_path = '<STR_LIT>' <EOL> self . fd = '<STR_LIT>' <EOL> def tearDown ( self ) : <EOL> self . mox . UnsetStubs ( ) <EOL> class FileUtilMoxTest ( FileUtilMoxTestBase ) : <EOL> def testListDirPath ( self ) : <EOL> self . mox . StubOutWithMock ( os , '<STR_LIT>' ) <EOL> dir_contents = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> os . listdir ( '<STR_LIT>' ) . AndReturn ( dir_contents ) <EOL> self . mox . ReplayAll ( ) <EOL> self . assertListEqual ( file_util . ListDirPath ( '<STR_LIT>' ) , <EOL> [ '<STR_LIT>' % ( '<STR_LIT>' , entry ) <EOL> for entry in dir_contents ] ) <EOL> self . mox . VerifyAll ( ) <EOL> def testSuccessfulRead ( self ) : <EOL> file_handle = self . mox . CreateMockAnything ( ) <EOL> self . mox . StubOutWithMock ( __builtin__ , '<STR_LIT>' , use_mock_anything = True ) <EOL> open ( self . file_path ) . AndReturn ( file_handle ) <EOL> file_handle . __enter__ ( ) . AndReturn ( file_handle ) <EOL> file_handle . read ( ) . AndReturn ( self . sample_contents ) <EOL> file_handle . __exit__ ( None , None , None ) <EOL> self . mox . ReplayAll ( ) <EOL> try : <EOL> self . assertEquals ( file_util . Read ( self . file_path ) , self . sample_contents ) <EOL> self . mox . VerifyAll ( ) <EOL> finally : <EOL> self . mox . ResetAll ( ) <EOL> self . mox . UnsetStubs ( ) <EOL> def testWriteGroup ( self ) : <EOL> self . mox . StubOutWithMock ( os , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( os , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( os , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( os , '<STR_LIT>' ) <EOL> gid = '<STR_LIT>' <EOL> os . open ( self . file_path , os . O_WRONLY | os . O_TRUNC | os . O_CREAT , <EOL> <NUM_LIT:0> <NUM_LIT> ) . AndReturn ( self . fd ) <EOL> os . write ( self . fd , self . sample_contents ) <EOL> os . close ( self . fd ) <EOL> os . chown ( self . file_path , - <NUM_LIT:1> , gid ) <EOL> self . mox . ReplayAll ( ) <EOL> file_util . Write ( self . file_path , self . sample_contents , gid = gid ) <EOL> self . mox . VerifyAll ( ) <EOL> class AtomicWriteMoxTest ( FileUtilMoxTestBase ) : <EOL> def setUp ( self ) : <EOL> super ( AtomicWriteMoxTest , self ) . setUp ( ) <EOL> self . mox . StubOutWithMock ( tempfile , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( os , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( os , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( os , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( os , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( os , '<STR_LIT>' ) <EOL> self . mode = '<STR_LIT>' <EOL> self . gid = '<STR_LIT>' <EOL> self . temp_filename = '<STR_LIT>' <EOL> self . os_error = OSError ( '<STR_LIT>' ) <EOL> tempfile . mkstemp ( dir = '<STR_LIT>' ) . AndReturn ( <EOL> ( self . fd , self . temp_filename ) ) <EOL> os . write ( self . fd , self . sample_contents ) <EOL> os . close ( self . fd ) <EOL> os . chmod ( self . temp_filename , self . mode ) <EOL> def tearDown ( self ) : <EOL> self . mox . UnsetStubs ( ) <EOL> def testAtomicWriteGroup ( self ) : <EOL> self . mox . StubOutWithMock ( os , '<STR_LIT>' ) <EOL> os . chown ( self . temp_filename , - <NUM_LIT:1> , self . gid ) <EOL> os . rename ( self . temp_filename , self . file_path ) <EOL> self . mox . ReplayAll ( ) <EOL> file_util . AtomicWrite ( self . file_path , self . sample_contents , <EOL> mode = self . mode , gid = self . gid ) <EOL> self . mox . VerifyAll ( ) <EOL> def testAtomicWriteGroupError ( self ) : <EOL> self . mox . StubOutWithMock ( os , '<STR_LIT>' ) <EOL> os . chown ( self . temp_filename , - <NUM_LIT:1> , self . gid ) . AndRaise ( self . os_error ) <EOL> os . remove ( self . temp_filename ) <EOL> self . mox . ReplayAll ( ) <EOL> self . assertRaises ( OSError , file_util . AtomicWrite , self . file_path , <EOL> self . sample_contents , mode = self . mode , gid = self . gid ) <EOL> self . mox . VerifyAll ( ) <EOL> def testRenamingError ( self ) : <EOL> os . rename ( self . temp_filename , self . file_path ) . AndRaise ( self . os_error ) <EOL> os . remove ( self . temp_filename ) <EOL> self . mox . ReplayAll ( ) <EOL> self . assertRaises ( OSError , file_util . AtomicWrite , self . file_path , <EOL> self . sample_contents , mode = self . mode ) <EOL> self . mox . VerifyAll ( ) <EOL> def testRenamingErrorWithRemoveError ( self ) : <EOL> extra_error = OSError ( '<STR_LIT>' ) <EOL> os . rename ( self . temp_filename , self . file_path ) . AndRaise ( self . os_error ) <EOL> os . remove ( self . temp_filename ) . AndRaise ( extra_error ) <EOL> self . mox . ReplayAll ( ) <EOL> try : <EOL> file_util . AtomicWrite ( self . file_path , self . sample_contents , <EOL> mode = self . mode ) <EOL> except OSError as e : <EOL> self . assertEquals ( str ( e ) , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> else : <EOL> raise self . failureException ( '<STR_LIT>' ) <EOL> self . mox . VerifyAll ( ) <EOL> class TemporaryFilesMoxTest ( FileUtilMoxTestBase ) : <EOL> def testTemporaryFileWithContents ( self ) : <EOL> contents = '<STR_LIT>' <EOL> with file_util . TemporaryFileWithContents ( contents ) as temporary_file : <EOL> filename = temporary_file . name <EOL> contents_read = open ( temporary_file . name ) . read ( ) <EOL> self . assertEqual ( contents_read , contents ) <EOL> self . assertFalse ( os . path . exists ( filename ) ) <EOL> class TemporaryDirsMoxTest ( FileUtilMoxTestBase ) : <EOL> def testTemporaryDirectoryWithException ( self ) : <EOL> def Inner ( accumulator ) : <EOL> with file_util . TemporaryDirectory ( base_path = FLAGS . test_tmpdir ) as tmpdir : <EOL> self . assertTrue ( os . path . isdir ( tmpdir ) ) <EOL> accumulator . append ( tmpdir ) <EOL> raise Exception ( '<STR_LIT>' ) <EOL> temp_dirs = [ ] <EOL> self . assertRaises ( Exception , Inner , temp_dirs ) <EOL> self . assertEquals ( len ( temp_dirs ) , <NUM_LIT:1> ) <EOL> self . assertFalse ( os . path . isdir ( temp_dirs [ <NUM_LIT:0> ] ) ) <EOL> def testTemporaryDirectory ( self ) : <EOL> with file_util . TemporaryDirectory ( base_path = FLAGS . test_tmpdir ) as temp_dir : <EOL> self . assertTrue ( os . path . isdir ( temp_dir ) ) <EOL> self . assertFalse ( os . path . isdir ( temp_dir ) ) <EOL> class MkDirsMoxTest ( FileUtilMoxTestBase ) : <EOL> def setUp ( self ) : <EOL> super ( MkDirsMoxTest , self ) . setUp ( ) <EOL> self . mox . StubOutWithMock ( os , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( os , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( os . path , '<STR_LIT>' ) <EOL> self . dir_tree = [ '<STR_LIT>' , '<STR_LIT:to>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> def tearDown ( self ) : <EOL> self . mox . UnsetStubs ( ) <EOL> def testNoErrorsAbsoluteOneDir ( self ) : <EOL> os . mkdir ( '<STR_LIT>' ) <EOL> self . mox . ReplayAll ( ) <EOL> file_util . MkDirs ( '<STR_LIT>' ) <EOL> self . mox . VerifyAll ( ) <EOL> def testNoErrorsAbsoluteOneDirWithForceMode ( self ) : <EOL> os . mkdir ( '<STR_LIT>' ) <EOL> os . chmod ( '<STR_LIT>' , <NUM_LIT:0> <NUM_LIT> ) <EOL> self . mox . ReplayAll ( ) <EOL> file_util . MkDirs ( '<STR_LIT>' , force_mode = <NUM_LIT:0> <NUM_LIT> ) <EOL> self . mox . VerifyAll ( ) <EOL> def testNoErrorsExistingDirWithForceMode ( self ) : <EOL> exist_error = OSError ( errno . EEXIST , '<STR_LIT>' ) <EOL> os . mkdir ( '<STR_LIT>' ) . AndRaise ( exist_error ) <EOL> os . path . isdir ( '<STR_LIT>' ) . AndReturn ( True ) <EOL> self . mox . ReplayAll ( ) <EOL> file_util . MkDirs ( '<STR_LIT>' , force_mode = <NUM_LIT:0> <NUM_LIT> ) <EOL> self . mox . VerifyAll ( ) <EOL> def testNoErrorsAbsoluteSlashDot ( self ) : <EOL> os . mkdir ( '<STR_LIT>' ) <EOL> self . mox . ReplayAll ( ) <EOL> file_util . MkDirs ( '<STR_LIT>' ) <EOL> self . mox . VerifyAll ( ) <EOL> def testNoErrorsAbsoluteExcessiveSlashDot ( self ) : <EOL> """<STR_LIT>""" <EOL> os . mkdir ( '<STR_LIT>' ) <EOL> os . mkdir ( '<STR_LIT>' ) <EOL> self . mox . ReplayAll ( ) <EOL> file_util . MkDirs ( '<STR_LIT>' ) <EOL> self . mox . VerifyAll ( ) <EOL> def testNoErrorsAbsoluteTwoDirs ( self ) : <EOL> os . mkdir ( '<STR_LIT>' ) <EOL> os . mkdir ( '<STR_LIT>' ) <EOL> self . mox . ReplayAll ( ) <EOL> file_util . MkDirs ( '<STR_LIT>' ) <EOL> self . mox . VerifyAll ( ) <EOL> def testNoErrorsPartialTwoDirsWithForceMode ( self ) : <EOL> exist_error = OSError ( errno . EEXIST , '<STR_LIT>' ) <EOL> os . mkdir ( '<STR_LIT>' ) . AndRaise ( exist_error ) <EOL> os . path . isdir ( '<STR_LIT>' ) . AndReturn ( True ) <EOL> os . mkdir ( '<STR_LIT>' ) <EOL> os . chmod ( '<STR_LIT>' , <NUM_LIT:0> <NUM_LIT> ) <EOL> self . mox . ReplayAll ( ) <EOL> file_util . MkDirs ( '<STR_LIT>' , force_mode = <NUM_LIT:0> <NUM_LIT> ) <EOL> self . mox . VerifyAll ( ) <EOL> def testNoErrorsRelativeOneDir ( self ) : <EOL> os . mkdir ( '<STR_LIT:foo>' ) <EOL> self . mox . ReplayAll ( ) <EOL> file_util . MkDirs ( '<STR_LIT:foo>' ) <EOL> self . mox . VerifyAll ( ) <EOL> def testNoErrorsRelativeTwoDirs ( self ) : <EOL> os . mkdir ( '<STR_LIT:foo>' ) <EOL> os . mkdir ( '<STR_LIT>' ) <EOL> self . mox . ReplayAll ( ) <EOL> file_util . MkDirs ( '<STR_LIT>' ) <EOL> self . mox . VerifyAll ( ) <EOL> def testDirectoriesExist ( self ) : <EOL> exist_error = OSError ( errno . EEXIST , '<STR_LIT>' ) <EOL> for i in range ( len ( self . dir_tree ) ) : <EOL> path = os . path . join ( * self . dir_tree [ : i + <NUM_LIT:1> ] ) <EOL> os . mkdir ( path ) . AndRaise ( exist_error ) <EOL> os . path . isdir ( path ) . AndReturn ( True ) <EOL> self . mox . ReplayAll ( ) <EOL> file_util . MkDirs ( os . path . join ( * self . dir_tree ) ) <EOL> self . mox . VerifyAll ( ) <EOL> def testFileInsteadOfDirectory ( self ) : <EOL> exist_error = OSError ( errno . EEXIST , '<STR_LIT>' ) <EOL> path = self . dir_tree [ <NUM_LIT:0> ] <EOL> os . mkdir ( path ) . AndRaise ( exist_error ) <EOL> os . path . isdir ( path ) . AndReturn ( False ) <EOL> self . mox . ReplayAll ( ) <EOL> self . assertRaises ( OSError , file_util . MkDirs , os . path . join ( * self . dir_tree ) ) <EOL> self . mox . VerifyAll ( ) <EOL> def testNonExistsError ( self ) : <EOL> non_exist_error = OSError ( errno . ETIMEDOUT , '<STR_LIT>' ) <EOL> path = self . dir_tree [ <NUM_LIT:0> ] <EOL> os . mkdir ( path ) . AndRaise ( non_exist_error ) <EOL> self . mox . ReplayAll ( ) <EOL> self . assertRaises ( OSError , file_util . MkDirs , os . path . join ( * self . dir_tree ) ) <EOL> self . mox . VerifyAll ( ) <EOL> class RmDirsTestCase ( mox . MoxTestBase ) : <EOL> def testRmDirs ( self ) : <EOL> test_sandbox = os . path . join ( FLAGS . test_tmpdir , '<STR_LIT>' ) <EOL> test_dir = os . path . join ( test_sandbox , '<STR_LIT:test>' , '<STR_LIT>' ) <EOL> os . makedirs ( test_sandbox ) <EOL> with open ( os . path . join ( test_sandbox , '<STR_LIT:file>' ) , '<STR_LIT:w>' ) : <EOL> pass <EOL> os . makedirs ( test_dir ) <EOL> with open ( os . path . join ( test_dir , '<STR_LIT:file>' ) , '<STR_LIT:w>' ) : <EOL> pass <EOL> file_util . RmDirs ( test_dir ) <EOL> self . assertFalse ( os . path . exists ( os . path . join ( test_sandbox , '<STR_LIT:test>' ) ) ) <EOL> self . assertTrue ( os . path . exists ( os . path . join ( test_sandbox , '<STR_LIT:file>' ) ) ) <EOL> shutil . rmtree ( test_sandbox ) <EOL> def testRmDirsForNonExistingDirectory ( self ) : <EOL> self . mox . StubOutWithMock ( os , '<STR_LIT>' ) <EOL> os . rmdir ( '<STR_LIT>' ) <EOL> os . rmdir ( '<STR_LIT:path>' ) <EOL> self . mox . StubOutWithMock ( shutil , '<STR_LIT>' ) <EOL> shutil . rmtree ( '<STR_LIT>' ) . AndRaise ( <EOL> OSError ( errno . ENOENT , "<STR_LIT>" ) ) <EOL> self . mox . ReplayAll ( ) <EOL> file_util . RmDirs ( '<STR_LIT>' ) <EOL> self . mox . VerifyAll ( ) <EOL> def testRmDirsForNonExistingParentDirectory ( self ) : <EOL> self . mox . StubOutWithMock ( os , '<STR_LIT>' ) <EOL> os . rmdir ( '<STR_LIT>' ) . AndRaise ( <EOL> OSError ( errno . ENOENT , "<STR_LIT>" ) ) <EOL> os . rmdir ( '<STR_LIT:path>' ) <EOL> self . mox . StubOutWithMock ( shutil , '<STR_LIT>' ) <EOL> shutil . rmtree ( '<STR_LIT>' ) . AndRaise ( <EOL> OSError ( errno . ENOENT , "<STR_LIT>" ) ) <EOL> self . mox . ReplayAll ( ) <EOL> file_util . RmDirs ( '<STR_LIT>' ) <EOL> self . mox . VerifyAll ( ) <EOL> def testRmDirsForNotEmptyDirectory ( self ) : <EOL> self . mox . StubOutWithMock ( os , '<STR_LIT>' ) <EOL> os . rmdir ( '<STR_LIT>' ) . AndRaise ( <EOL> OSError ( errno . ENOTEMPTY , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . mox . StubOutWithMock ( shutil , '<STR_LIT>' ) <EOL> shutil . rmtree ( '<STR_LIT>' ) <EOL> self . mox . ReplayAll ( ) <EOL> file_util . RmDirs ( '<STR_LIT>' ) <EOL> self . mox . VerifyAll ( ) <EOL> def testRmDirsForPermissionDeniedOnParentDirectory ( self ) : <EOL> self . mox . StubOutWithMock ( os , '<STR_LIT>' ) <EOL> os . rmdir ( '<STR_LIT>' ) . AndRaise ( <EOL> OSError ( errno . EACCES , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . mox . StubOutWithMock ( shutil , '<STR_LIT>' ) <EOL> shutil . rmtree ( '<STR_LIT>' ) <EOL> self . mox . ReplayAll ( ) <EOL> file_util . RmDirs ( '<STR_LIT>' ) <EOL> self . mox . VerifyAll ( ) <EOL> def testRmDirsWithSimplePath ( self ) : <EOL> self . mox . StubOutWithMock ( shutil , '<STR_LIT>' ) <EOL> shutil . rmtree ( '<STR_LIT>' ) <EOL> self . mox . ReplayAll ( ) <EOL> file_util . RmDirs ( '<STR_LIT>' ) <EOL> self . mox . VerifyAll ( ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> basetest . main ( ) </s>
