<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import <EOL> import os <EOL> import gslib . tests . testcase as testcase <EOL> from gslib . tests . util import ObjectToURI as suri <EOL> from gslib . tests . util import SequentialAndParallelTransfer <EOL> from gslib . util import Retry <EOL> class TestMv ( testcase . GsUtilIntegrationTestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_moving ( self ) : <EOL> """<STR_LIT>""" <EOL> bucket1_uri = self . CreateBucket ( test_objects = <NUM_LIT:2> ) <EOL> self . AssertNObjectsInBucket ( bucket1_uri , <NUM_LIT:2> ) <EOL> bucket2_uri = self . CreateBucket ( ) <EOL> self . AssertNObjectsInBucket ( bucket2_uri , <NUM_LIT:0> ) <EOL> objs = [ bucket1_uri . clone_replace_key ( key ) . versionless_uri <EOL> for key in bucket1_uri . list_bucket ( ) ] <EOL> cmd = ( [ '<STR_LIT>' , '<STR_LIT>' ] + objs + [ suri ( bucket2_uri ) ] ) <EOL> stderr = self . RunGsUtil ( cmd , return_stderr = True ) <EOL> self . assertGreaterEqual ( stderr . count ( '<STR_LIT>' ) , <NUM_LIT:2> ) <EOL> self . assertLessEqual ( stderr . count ( '<STR_LIT>' ) , <NUM_LIT:4> ) <EOL> self . assertEqual ( stderr . count ( '<STR_LIT>' ) % <NUM_LIT:2> , <NUM_LIT:0> ) <EOL> self . assertEqual ( stderr . count ( '<STR_LIT>' ) , <NUM_LIT:2> ) <EOL> self . AssertNObjectsInBucket ( bucket1_uri , <NUM_LIT:0> ) <EOL> self . AssertNObjectsInBucket ( bucket2_uri , <NUM_LIT:2> ) <EOL> objs = [ bucket2_uri . clone_replace_key ( key ) . versionless_uri <EOL> for key in bucket2_uri . list_bucket ( ) ] <EOL> obj1 = objs [ <NUM_LIT:0> ] <EOL> self . RunGsUtil ( [ '<STR_LIT>' , obj1 ] ) <EOL> self . AssertNObjectsInBucket ( bucket1_uri , <NUM_LIT:0> ) <EOL> self . AssertNObjectsInBucket ( bucket2_uri , <NUM_LIT:1> ) <EOL> objs = [ suri ( bucket2_uri . clone_replace_key ( key ) ) <EOL> for key in bucket2_uri . list_bucket ( ) ] <EOL> cmd = ( [ '<STR_LIT>' , '<STR_LIT>' ] + objs + [ suri ( bucket1_uri ) ] ) <EOL> stderr = self . RunGsUtil ( cmd , return_stderr = True ) <EOL> self . assertGreaterEqual ( stderr . count ( '<STR_LIT>' ) , <NUM_LIT:1> ) <EOL> self . assertLessEqual ( stderr . count ( '<STR_LIT>' ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( stderr . count ( '<STR_LIT>' ) , <NUM_LIT:1> ) <EOL> self . AssertNObjectsInBucket ( bucket1_uri , <NUM_LIT:1> ) <EOL> self . AssertNObjectsInBucket ( bucket2_uri , <NUM_LIT:0> ) <EOL> def test_move_dir_to_bucket ( self ) : <EOL> """<STR_LIT>""" <EOL> bucket_uri = self . CreateBucket ( ) <EOL> dir_to_move = self . CreateTempDir ( test_files = <NUM_LIT:2> ) <EOL> self . RunGsUtil ( [ '<STR_LIT>' , dir_to_move , suri ( bucket_uri ) ] ) <EOL> self . AssertNObjectsInBucket ( bucket_uri , <NUM_LIT:2> ) <EOL> @ SequentialAndParallelTransfer <EOL> def test_stdin_args ( self ) : <EOL> """<STR_LIT>""" <EOL> tmpdir = self . CreateTempDir ( ) <EOL> fpath1 = self . CreateTempFile ( tmpdir = tmpdir , contents = '<STR_LIT>' ) <EOL> fpath2 = self . CreateTempFile ( tmpdir = tmpdir , contents = '<STR_LIT>' ) <EOL> bucket_uri = self . CreateBucket ( ) <EOL> self . RunGsUtil ( [ '<STR_LIT>' , '<STR_LIT>' , suri ( bucket_uri ) ] , <EOL> stdin = '<STR_LIT:\n>' . join ( ( fpath1 , fpath2 ) ) ) <EOL> @ Retry ( AssertionError , tries = <NUM_LIT:3> , timeout_secs = <NUM_LIT:1> ) <EOL> def _Check1 ( ) : <EOL> stdout = self . RunGsUtil ( [ '<STR_LIT>' , suri ( bucket_uri ) ] , return_stdout = True ) <EOL> self . assertIn ( os . path . basename ( fpath1 ) , stdout ) <EOL> self . assertIn ( os . path . basename ( fpath2 ) , stdout ) <EOL> self . assertNumLines ( stdout , <NUM_LIT:2> ) <EOL> _Check1 ( ) <EOL> def test_mv_no_clobber ( self ) : <EOL> """<STR_LIT>""" <EOL> fpath1 = self . CreateTempFile ( contents = '<STR_LIT>' ) <EOL> bucket_uri = self . CreateBucket ( ) <EOL> object_uri = self . CreateObject ( bucket_uri = bucket_uri , contents = '<STR_LIT>' ) <EOL> stderr = self . RunGsUtil ( [ '<STR_LIT>' , '<STR_LIT>' , fpath1 , suri ( object_uri ) ] , <EOL> return_stderr = True ) <EOL> self . assertIn ( '<STR_LIT>' % suri ( object_uri ) , stderr ) <EOL> self . assertNotIn ( '<STR_LIT>' % suri ( fpath1 ) , stderr ) <EOL> contents = self . RunGsUtil ( [ '<STR_LIT>' , suri ( object_uri ) ] , return_stdout = True ) <EOL> self . assertEqual ( contents , '<STR_LIT>' ) </s>
