make gs_utils_manualtest use a test-only bucket for safety TBR=borenet Review URL: https://codereview.chromium.org/412553004
diff --git a/py/utils/gs_utils_manualtest.py b/py/utils/gs_utils_manualtest.py index 6751647..db70267 100755 --- a/py/utils/gs_utils_manualtest.py +++ b/py/utils/gs_utils_manualtest.py
@@ -16,27 +16,37 @@ # Local imports. import gs_utils - -def _test_public_read(): - """Make sure we can read from public files without .boto file credentials.""" - gs = gs_utils.GSUtils() - gs.list_bucket_contents(bucket='chromium-skia-gm-summaries', subdir=None) +TEST_BUCKET = 'chromium-skia-testing' -def _test_authenticated_round_trip(): +def _get_authenticated_gs_handle(): + """Returns an instance of GSUtils using ~/.boto for authentication.""" try: - gs = gs_utils.GSUtils( + return gs_utils.GSUtils( boto_file_path=os.path.expanduser(os.path.join('~','.boto'))) except: print """ Failed to instantiate GSUtils object with default .boto file path. -Do you have a ~/.boto file that provides the credentials needed to read -and write gs://chromium-skia-gm ? -""" +Do you have a ~/.boto file that provides the credentials needed to write +into gs://%s? +""" % TEST_BUCKET raise - bucket = 'chromium-skia-gm' - remote_dir = 'gs_utils_test/%d' % random.randint(0, sys.maxint) + +def _get_unique_posix_dir(): + """Returns a unique directory name suitable for use in Google Storage.""" + return 'gs_utils_manualtest/%d' % random.randint(0, sys.maxint) + + +def _test_public_read(): + """Make sure we can read from public files without .boto file credentials.""" + gs = gs_utils.GSUtils() + gs.list_bucket_contents(bucket=TEST_BUCKET, subdir=None) + + +def _test_authenticated_round_trip(): + gs = _get_authenticated_gs_handle() + remote_dir = _get_unique_posix_dir() subdir = 'subdir' filenames_to_upload = ['file1', 'file2'] @@ -54,9 +64,9 @@ dest_path = posixpath.join(remote_dir, subdir, filename) gs.upload_file( source_path=os.path.join(local_src_dir, subdir, filename), - dest_bucket=bucket, dest_path=dest_path, + dest_bucket=TEST_BUCKET, dest_path=dest_path, fine_grained_acl_list=[(id_type, id_value, set_permission)]) - got_permission = gs.get_acl(bucket=bucket, path=dest_path, + got_permission = gs.get_acl(bucket=TEST_BUCKET, path=dest_path, id_type=id_type, id_value=id_value) assert got_permission == set_permission, '%s == %s' % ( got_permission, set_permission) @@ -65,11 +75,11 @@ # Get a list of the files we uploaded to Google Storage. (dirs, files) = gs.list_bucket_contents( - bucket=bucket, subdir=remote_dir) + bucket=TEST_BUCKET, subdir=remote_dir) assert dirs == [subdir], '%s == [%s]' % (dirs, subdir) assert files == [], '%s == []' % files (dirs, files) = gs.list_bucket_contents( - bucket=bucket, subdir=posixpath.join(remote_dir, subdir)) + bucket=TEST_BUCKET, subdir=posixpath.join(remote_dir, subdir)) assert dirs == [], '%s == []' % dirs assert files == filenames_to_upload, '%s == %s' % (files, filenames_to_upload) @@ -80,30 +90,30 @@ id_value = 'google.com' fullpath = posixpath.join(remote_dir, subdir, filenames_to_upload[0]) # Make sure ACL is empty to start with ... - gs.set_acl(bucket=bucket, path=fullpath, + gs.set_acl(bucket=TEST_BUCKET, path=fullpath, id_type=id_type, id_value=id_value, permission=gs.Permission.EMPTY) - permission = gs.get_acl(bucket=bucket, path=fullpath, + permission = gs.get_acl(bucket=TEST_BUCKET, path=fullpath, id_type=id_type, id_value=id_value) assert permission == gs.Permission.EMPTY, '%s == %s' % ( permission, gs.Permission.EMPTY) # ... set it to OWNER ... - gs.set_acl(bucket=bucket, path=fullpath, + gs.set_acl(bucket=TEST_BUCKET, path=fullpath, id_type=id_type, id_value=id_value, permission=gs.Permission.OWNER) - permission = gs.get_acl(bucket=bucket, path=fullpath, + permission = gs.get_acl(bucket=TEST_BUCKET, path=fullpath, id_type=id_type, id_value=id_value) assert permission == gs.Permission.OWNER, '%s == %s' % ( permission, gs.Permission.OWNER) # ... now set it to READ ... - gs.set_acl(bucket=bucket, path=fullpath, + gs.set_acl(bucket=TEST_BUCKET, path=fullpath, id_type=id_type, id_value=id_value, permission=gs.Permission.READ) - permission = gs.get_acl(bucket=bucket, path=fullpath, + permission = gs.get_acl(bucket=TEST_BUCKET, path=fullpath, id_type=id_type, id_value=id_value) assert permission == gs.Permission.READ, '%s == %s' % ( permission, gs.Permission.READ) # ... and clear it again to finish. - gs.set_acl(bucket=bucket, path=fullpath, + gs.set_acl(bucket=TEST_BUCKET, path=fullpath, id_type=id_type, id_value=id_value, permission=gs.Permission.EMPTY) - permission = gs.get_acl(bucket=bucket, path=fullpath, + permission = gs.get_acl(bucket=TEST_BUCKET, path=fullpath, id_type=id_type, id_value=id_value) assert permission == gs.Permission.EMPTY, '%s == %s' % ( permission, gs.Permission.EMPTY) @@ -112,7 +122,7 @@ local_dest_dir = tempfile.mkdtemp() try: for filename in filenames_to_upload: - gs.download_file(source_bucket=bucket, + gs.download_file(source_bucket=TEST_BUCKET, source_path=posixpath.join(remote_dir, subdir, filename), dest_path=os.path.join(local_dest_dir, subdir, filename), create_subdirs_if_needed=True) @@ -125,31 +135,20 @@ # Delete all the files we uploaded to Google Storage. for filename in filenames_to_upload: - gs.delete_file(bucket=bucket, + gs.delete_file(bucket=TEST_BUCKET, path=posixpath.join(remote_dir, subdir, filename)) # Confirm that we deleted all the files we uploaded to Google Storage. (dirs, files) = gs.list_bucket_contents( - bucket=bucket, subdir=posixpath.join(remote_dir, subdir)) + bucket=TEST_BUCKET, subdir=posixpath.join(remote_dir, subdir)) assert dirs == [], '%s == []' % dirs assert files == [], '%s == []' % files def _test_dir_upload_and_download(): """Test upload_dir_contents() and download_dir_contents().""" - try: - gs = gs_utils.GSUtils( - boto_file_path=os.path.expanduser(os.path.join('~','.boto'))) - except: - print """ -Failed to instantiate GSUtils object with default .boto file path. -Do you have a ~/.boto file that provides the credentials needed to read -and write gs://chromium-skia-gm ? -""" - raise - - bucket = 'chromium-skia-gm' - remote_dir = 'gs_utils_test/%d' % random.randint(0, sys.maxint) + gs = _get_authenticated_gs_handle() + remote_dir = _get_unique_posix_dir() subdir = 'subdir' filenames = ['file1', 'file2'] @@ -164,7 +163,7 @@ with open(os.path.join(local_src_dir, subdir, filename), 'w') as f: f.write('contents of %s\n' % filename) gs.upload_dir_contents( - source_dir=local_src_dir, dest_bucket=bucket, dest_dir=remote_dir, + source_dir=local_src_dir, dest_bucket=TEST_BUCKET, dest_dir=remote_dir, predefined_acl=gs.PredefinedACL.PRIVATE, fine_grained_acl_list=[(id_type, id_value, set_permission)]) finally: @@ -172,18 +171,18 @@ # Validate the list of the files we uploaded to Google Storage. (dirs, files) = gs.list_bucket_contents( - bucket=bucket, subdir=remote_dir) + bucket=TEST_BUCKET, subdir=remote_dir) assert dirs == [subdir], '%s == [%s]' % (dirs, subdir) assert files == [], '%s == []' % files (dirs, files) = gs.list_bucket_contents( - bucket=bucket, subdir=posixpath.join(remote_dir, subdir)) + bucket=TEST_BUCKET, subdir=posixpath.join(remote_dir, subdir)) assert dirs == [], '%s == []' % dirs assert files == filenames, '%s == %s' % (files, filenames) # Check the fine-grained ACLs we set in Google Storage. for filename in filenames: got_permission = gs.get_acl( - bucket=bucket, path=posixpath.join(remote_dir, subdir, filename), + bucket=TEST_BUCKET, path=posixpath.join(remote_dir, subdir, filename), id_type=id_type, id_value=id_value) assert got_permission == set_permission, '%s == %s' % ( got_permission, set_permission) @@ -192,7 +191,7 @@ # are what we expect, and then delete the tree in Google Storage. local_dest_dir = tempfile.mkdtemp() try: - gs.download_dir_contents(source_bucket=bucket, source_dir=remote_dir, + gs.download_dir_contents(source_bucket=TEST_BUCKET, source_dir=remote_dir, dest_dir=local_dest_dir) for filename in filenames: with open(os.path.join(local_dest_dir, subdir, filename)) as f: @@ -202,7 +201,7 @@ finally: shutil.rmtree(local_dest_dir) for filename in filenames: - gs.delete_file(bucket=bucket, + gs.delete_file(bucket=TEST_BUCKET, path=posixpath.join(remote_dir, subdir, filename))