Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_increase_verbosity(self):
"""Make sure increase_verbosity() respects default and custom levels."""
# Start from a known state.
set_level(logging.INFO)
assert get_level() == logging.INFO
# INFO -> VERBOSE.
increase_verbosity()
assert get_level() == logging.VERBOSE
# VERBOSE -> DEBUG.
increase_verbosity()
assert get_level() == logging.DEBUG
# DEBUG -> SPAM.
increase_verbosity()
assert get_level() == logging.SPAM
# SPAM -> NOTSET.
increase_verbosity()
assert get_level() == logging.NOTSET
# NOTSET -> NOTSET.
increase_verbosity()
assert get_level() == logging.NOTSET
1. Sets up verbose logging to the terminal. When a test fails the logging
output can help to perform a post-mortem analysis of the failure in
question (even when its hard to reproduce locally). This is especially
useful when debugging remote test failures, whether they happened on
Travis CI or a user's local system.
2. Creates temporary directories where the pip download cache and the
pip-accel binary cache are located. Isolating the pip-accel binary cache
from the user's system is meant to ensure that the tests are as
independent from the user's system as possible. The function
:func:`tearDownModule` is responsible for cleaning up the temporary
directory after the test suite finishes.
"""
# Initialize verbose logging to the terminal.
coloredlogs.install()
coloredlogs.increase_verbosity()
# Create temporary directories to store the pip download cache and
# pip-accel's binary cache, to make sure these tests run isolated from the
# rest of the system.
os.environ['PIP_DOWNLOAD_CACHE'] = create_temporary_directory()
os.environ['PIP_ACCEL_CACHE'] = create_temporary_directory()
'exclusive', 'lock-timeout=', 'lock-file=', 'timeout=',
'fudge-factor=', 'verbose', 'quiet', 'help',
])
for option, value in options:
if option in ('-e', '--exclusive'):
exclusive = True
elif option in ('-T', '--lock-timeout'):
lock_timeout = parse_timespan(value)
elif option in ('-l', '--lock-file'):
lock_name = value
elif option in ('-t', '--timeout'):
command_timeout = parse_timespan(value)
elif option in ('-f', '--fudge-factor'):
fudge_factor = parse_timespan(value)
elif option in ('-v', '--verbose'):
coloredlogs.increase_verbosity()
elif option in ('-q', '--quiet'):
coloredlogs.decrease_verbosity()
elif option in ('-h', '--help'):
usage(__doc__)
sys.exit(0)
else:
assert False, "Unhandled option!"
# Make sure the operator provided a program to execute.
if not arguments:
usage(__doc__)
sys.exit(0)
# Make sure the program actually exists.
program_name = arguments[0]
if not os.path.isfile(program_name):
# Only search the $PATH if the given program name
# doesn't already include one or more path segments.
# to pip without any changes and exit immediately afterwards.
if 'install' not in arguments:
# This will not return.
os.execvp('pip', ['pip'] + arguments)
else:
arguments = [arg for arg in arguments if arg != 'install']
config = Config()
# Initialize logging output.
coloredlogs.install(
fmt=config.log_format,
level=config.log_verbosity,
)
# Adjust verbosity based on -v, -q, --verbose, --quiet options.
for argument in list(arguments):
if match_option(argument, '-v', '--verbose'):
coloredlogs.increase_verbosity()
elif match_option(argument, '-q', '--quiet'):
coloredlogs.decrease_verbosity()
# Perform the requested action(s).
try:
accelerator = PipAccelerator(config)
accelerator.install_from_arguments(arguments)
except NothingToDoError as e:
# Don't print a traceback for this (it's not very user friendly) and
# exit with status zero to stay compatible with pip. For more details
# please refer to https://github.com/paylogic/pip-accel/issues/47.
logger.warning("%s", e)
sys.exit(0)
except Exception:
logger.exception("Caught unhandled exception!")
sys.exit(1)
assert repository, "Please specify a repository first!"
actions.append(functools.partial(repository.update))
elif option in ('-m', '--merge-up'):
assert repository, "Please specify a repository first!"
actions.append(functools.partial(
repository.merge_up,
target_branch=revision,
feature_branch=arguments[0] if arguments else None,
))
elif option in ('-e', '--export'):
directory = value.strip()
assert repository, "Please specify a repository first!"
assert directory, "Please specify the directory where the revision should be exported!"
actions.append(functools.partial(repository.export, directory, revision))
elif option in ('-v', '--verbose'):
coloredlogs.increase_verbosity()
elif option in ('-q', '--quiet'):
coloredlogs.decrease_verbosity()
elif option in ('-h', '--help'):
usage(__doc__)
return
if not actions:
usage(__doc__)
return
except Exception as e:
warning("Error: %s", e)
sys.exit(1)
# Execute the requested action(s).
try:
for action in actions:
action()
except Exception:
elif option in ('-i', '--ionice'):
value = validate_ionice_class(value.lower().strip())
kw['io_scheduling_class'] = value
elif option in ('-c', '--config'):
kw['config_file'] = parse_path(value)
elif option in ('-u', '--use-sudo'):
use_sudo = True
elif option in ('-n', '--dry-run'):
logger.info("Performing a dry run (because of %s option) ..", option)
kw['dry_run'] = True
elif option in ('-C', '--removal-command'):
removal_command = shlex.split(value)
logger.info("Using custom removal command: %s", removal_command)
kw['removal_command'] = removal_command
elif option in ('-v', '--verbose'):
coloredlogs.increase_verbosity()
elif option in ('-q', '--quiet'):
coloredlogs.decrease_verbosity()
elif option in ('-h', '--help'):
usage(__doc__)
return
else:
assert False, "Unhandled option! (programming error)"
if rotation_scheme:
logger.verbose("Rotation scheme defined on command line: %s", rotation_scheme)
if arguments:
# Rotation of the locations given on the command line.
location_source = 'command line arguments'
selected_locations.extend(coerce_location(value, sudo=use_sudo) for value in arguments)
else:
# Rotation of all configured locations.
location_source = 'configuration file'
def rotate_database_dumps(dry_run=False):
"""Rotate database backups for real"""
try:
from executor.contexts import LocalContext, RemoteContext
from rotate_backups import RotateBackups, Location
import rotate_backups
import coloredlogs
except ImportError:
print(red("This fab command should be run within the venv."))
return
rotate_backups.TIMESTAMP_PATTERN = re.compile(
r'(?P\d{4})(?P\d{2})(?P\d{2})')
coloredlogs.increase_verbosity()
rotation_scheme = {
# same as doc/borg_backup_script/assembl_borg_backup.sh
'daily': 7, 'weekly': 4, 'monthly': 6,
# Plus yearly for good conscience
'yearly': 'always'
}
dir = env.dbdumps_dir
if running_locally([env.host_string]):
ctx = LocalContext()
dir = os.path.realpath(dir)
else:
ctx = RemoteContext(ssh_alias=env.host_string, ssh_user=env.user)
location = Location(context=ctx, directory=dir)
backup = RotateBackups(rotation_scheme, include_list=['db_*.sql.pgdump', 'db_*.bp'], dry_run=dry_run)
backup.rotate_backups(location, False)
if value.strip().startswith(('http://', 'https://', 'ftp://', 'mirror://')):
actions.append(functools.partial(updater.change_mirror, value))
else:
raise Exception("\'%s\' is not a valid mirror URL" % value)
elif option in ('-a', '--auto-change-mirror'):
actions.append(updater.change_mirror)
elif option in ('-u', '--update', '--update-package-lists'):
actions.append(updater.smart_update)
elif option in ('-U', '--ubuntu'):
ubuntu_mode = True
elif option in ('-x', '--exclude'):
actions.insert(0, functools.partial(updater.ignore_mirror, value))
elif option in ('-m', '--max'):
limit = int(value)
elif option in ('-v', '--verbose'):
coloredlogs.increase_verbosity()
elif option in ('-V', '--version'):
output("Version: %s on Python %i.%i", updater_version, sys.version_info[0], sys.version_info[1])
return
elif option in ('-C', '--codename'):
codename = value
elif option in ('-R', '--create-chroot'):
chroot_path = value
elif option in ('-q', '--quiet'):
coloredlogs.decrease_verbosity()
elif option in ('-h', '--help'):
usage(__doc__)
return
else:
assert False, "Unhandled option!"
if codename and not chroot_path:
assert False, "--codename must be used with valid -R to specify chroot path"
coloredlogs.install()
# Parse and validate the command line arguments.
try:
# Command line option defaults.
hostname = None
message = None
# Parse the command line options.
options, arguments = getopt.getopt(sys.argv[1:], 'b:n:m:vh',
['hostname=', 'message=', 'verbose', 'help'])
for option, value in options:
if option in ('-n', '--hostname'):
hostname = value
elif option in ('-m', '--message'):
message = value
elif option in ('-v', '--verbose'):
coloredlogs.increase_verbosity()
elif option in ('-h', '--help'):
usage()
return
else:
# Programming error...
assert False, "Unhandled option!"
# Handle the positional arguments.
if len(arguments) < 2:
usage()
return
supported_actions = ('start', 'commit', 'kill', 'delete')
action = arguments.pop(0)
if action not in supported_actions:
msg = "Action not supported: %r (supported actions are: %s)"
raise Exception, msg % (action, ', '.join(supported_actions))
except Exception, e: