Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def lock_is_released_by_context_manager_exit(lock_path):
has_run = False
# Keep a reference to first_lock so it holds onto the lock
first_lock = locket.lock_file(lock_path, timeout=0)
with first_lock:
pass
with locket.lock_file(lock_path, timeout=0):
has_run = True
assert has_run
def lock_can_be_acquired_with_timeout_of_zero(lock_path):
has_run = False
with locket.lock_file(lock_path, timeout=0):
has_run = True
assert has_run
def thread_cannot_obtain_lock_using_same_path_twice_without_release(lock_path):
with locket.lock_file(lock_path, timeout=0):
lock = locket.lock_file(lock_path, timeout=0)
try:
lock.acquire()
assert False, "Expected LockError"
except locket.LockError:
pass
def the_same_lock_file_object_is_used_for_the_same_path_with_different_arguments(lock_path):
# We explicitly check the same lock is used to ensure that the lock isn't
# re-entrant, even if the underlying platform lock is re-entrant.
first_lock = locket.lock_file(lock_path, timeout=None)
second_lock = locket.lock_file(lock_path, timeout=0)
assert first_lock._lock is second_lock._lock
def different_file_objects_are_used_for_different_paths(lock_path):
first_lock = locket.lock_file(lock_path, timeout=0)
second_lock = locket.lock_file(lock_path + "-2", timeout=0)
assert first_lock._lock is not second_lock._lock
def the_same_lock_file_object_is_used_for_the_same_path_with_different_arguments(lock_path):
# We explicitly check the same lock is used to ensure that the lock isn't
# re-entrant, even if the underlying platform lock is re-entrant.
first_lock = locket.lock_file(lock_path, timeout=None)
second_lock = locket.lock_file(lock_path, timeout=0)
assert first_lock._lock is second_lock._lock
def different_file_objects_are_used_for_different_paths(lock_path):
first_lock = locket.lock_file(lock_path, timeout=0)
second_lock = locket.lock_file(lock_path + "-2", timeout=0)
assert first_lock._lock is not second_lock._lock
def read_adc(self):
""" Read voltage from ADC """
try:
lock_acquired = False
# Set up lock for ADC
adc_lock = locket.lock_file(self.adc_lock_file, timeout=120)
try:
adc_lock.acquire()
lock_acquired = True
except:
self.logger.error("Could not acquire ADC lock. Breaking for future locking.")
os.remove(self.adc_lock_file)
if not lock_acquired:
self.logger.error(
"Unable to acquire lock: {lock}".format(
lock=self.adc_lock_file))
# Get measurement from ADC
measurements = self.adc.next()
if measurements is not None: