Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
class Git(git.cmd.Git):
"""Prevents asking for password for private repos"""
env = {'GIT_ASKPASS': 'echo'}
def __getattr__(self, item):
def wrapper(*args, **kwargs):
env = kwargs.pop('env', {})
env.update(self.env)
return super(Git, self).__getattr__(item)(*args, env=env, **kwargs)
return wrapper
class MsmProcessLock(InterProcessLock):
def __init__(self):
lock_path = '/tmp/msm_lock'
if not exists(lock_path):
lock_file = open(lock_path, '+w')
lock_file.close()
chmod(lock_path, 0o777)
super().__init__(lock_path)
# The cached_property class defined below was copied from the
# PythonDecoratorLibrary at:
# https://wiki.python.org/moin/PythonDecoratorLibrary/#Cached_Properties
#
# Β© 2011 Christopher Arndt, MIT License
#
class cached_property(object):
"""Decorator for read-only properties evaluated only once within TTL period.
# make sure to give it the absolute pathname to the program.
arguments[0] = program_name
except Exception as e:
warning("Failed to parse command line arguments: %s", e)
sys.exit(1)
# Apply the requested fudge factor.
apply_fudge_factor(fudge_factor)
# Run the requested command.
try:
if exclusive:
# Select a default lock file name?
if not lock_name:
lock_name = os.path.basename(arguments[0])
logger.debug("Using base name of command as lock file name (%s).", lock_name)
lock_file = get_lock_path(lock_name)
lock = InterProcessLock(path=lock_file, logger=logger)
logger.debug("Trying to acquire exclusive lock: %s", lock_file)
if lock.acquire(blocking=(lock_timeout > 0), max_delay=lock_timeout):
logger.info("Successfully acquired exclusive lock: %s", lock_file)
run_command(arguments, timeout=command_timeout)
else:
logger.error("Failed to acquire exclusive lock: %s", lock_file)
sys.exit(1)
else:
run_command(arguments, timeout=command_timeout)
except ExternalCommandFailed as e:
logger.error("%s", e.error_message)
sys.exit(e.command.returncode)
# We don't want to hose the machine, so even when collecting fails, we
# still wait for the configured interval.
update_stamp(collect_stamp_file)
try:
subprocess.check_output((bin_path('submit'),),
stderr=subprocess.STDOUT)
except subprocess.CalledProcessError as e:
log.error('submit failed:\n{}', e.output.decode('utf8'))
else:
update_stamp(submit_stamp_file)
if __name__ == '__main__':
lock_file = os.path.join(var_dir, 'client-cron.lock')
lock = InterProcessLock(lock_file)
# Note: Not necessary to explicitly release the lock, since fasteners
# guarantees that the lock is released when the process exits.
if not lock.acquire(timeout=59, delay=1, max_delay=1):
log.warn('Failed to acquire lock')
sys.exit(1)
main()
def __init__(self, path):
# Create lock file if it doesn't exist and set permissions for
# all users to lock/unlock
if not exists(path):
f = open(path, 'w+')
f.close()
chmod(path, 0o777)
self.plock = InterProcessLock(path)
self.tlock = Lock()
def __init__(self, data_file, start_time):
assert isinstance(data_file, str) and os.path.isfile(data_file)
assert isinstance(start_time, float)
self._lock = fasteners.process_lock.InterProcessLock("%s.lock" % (data_file,))
self.data_file = data_file
self.ignored = 0
self.iteration = 0
self.log_size = 0
self.results = 0
self.start_time = start_time
self.test_name = None
self.timestamp = start_time