blob: 52a4774e65120f4c8baa0e4a91d8b60dc984157b [file] [log] [blame]
# Copyright 2015 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Testable functions for Bucket."""
import logging
import os
import sys
import subprocess
from infra.path_hacks.depot_tools import _depot_tools as depot_tools_path
# https://chromium.googlesource.com/infra/infra/+/master/infra_libs/logs/README.md
LOGGER = logging.getLogger(__name__)
PROJECT = 'chromium-archive'
class BucketExists(Exception):
pass
class InvalidBucketName(Exception):
pass
def gsutil(args): # pragma: no cover
target = os.path.join(depot_tools_path, 'gsutil.py')
cmd = [sys.executable, target, '--']
cmd.extend(args)
print 'gsutil',
print ' '.join(args)
return subprocess.check_call(cmd, stderr=subprocess.PIPE)
def add_argparse_options(parser):
"""Define command-line arguments."""
parser.add_argument('bucket', type=str, nargs='+')
parser.add_argument(
'--reader',
'-r',
type=str,
action='append',
default=[],
help='Add this account as a Storage Legacy Bucket Reader')
parser.add_argument(
'--writer',
'-w',
type=str,
action='append',
default=[],
help='Add this account as a Storage Legacy Bucket Writer')
def ensure_no_bucket_exists(bucket):
"""Raises an exception if the bucket exists."""
try:
gsutil(['ls', '-b', 'gs://%s' % bucket])
except subprocess.CalledProcessError:
return
raise BucketExists('%s already exists.' % bucket)
def bucket_is_public(bucket_name):
"""Verify the name of the bucket and return whether it's public or not."""
if bucket_name.startswith('chromium-'):
return True
elif bucket_name.startswith('chrome-'):
return False
else:
raise InvalidBucketName(
'%s does not start with either "chromium-" or "chrome-"' % bucket_name)
def run(bucket_name, readers, writers, public):
ensure_no_bucket_exists(bucket_name)
gsutil(['mb', '-p', PROJECT, 'gs://%s' % bucket_name])
for reader in readers:
gsutil(['acl', 'ch', '-u', '%s:r' % reader, 'gs://%s' % bucket_name])
for writer in writers:
gsutil(['acl', 'ch', '-u', '%s:w' % writer, 'gs://%s' % bucket_name])
if public:
reader = 'AllUsers'
else:
reader = 'google.com'
gsutil(['acl', 'ch', '-g', '%s:R' % reader, 'gs://%s' % bucket_name])
gsutil(['defacl', 'ch', '-g', '%s:R' % reader, 'gs://%s' % bucket_name])