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_write_req15(self) -> None:
"""
# Given: a Downgradable RW lock type to instantiate a RW lock.
# When: A a generated writer lock is downgraed but it wasn't in a locked state.
# Then: the generated locks raise an exception if released while being unlocked.
"""
# ## Arrange
for current_rw_lock_type in self.c_rwlock_type_downgradable:
with self.subTest(current_rw_lock_type):
current_rw_lock = current_rw_lock_type()
assert isinstance(current_rw_lock, rwlock.RWLockableD)
current_lock: Union[rwlock.LockableD] = current_rw_lock.gen_wlock()
with self.assertRaises(rwlock.RELEASE_ERR_CLS) as err:
current_lock.release()
self.assertEqual(str(err.exception), str(rwlock.RELEASE_ERR_MSG))
with self.assertRaises(rwlock.RELEASE_ERR_CLS):
# ## Assume
self.assertFalse((current_lock.locked()))
# ## Act
current_lock.downgrade()
self.assertFalse(current_lock.locked())
self.c_rw_lock.c_lock_read.release()
def locked(self) -> bool:
"""Answer to 'is it currently locked?'."""
return self.v_locked
def gen_rlock(self) -> "RWLockFair._aReader":
"""Generate a reader lock."""
return RWLockFair._aReader(self)
def gen_wlock(self) -> "RWLockFair._aWriter":
"""Generate a writer lock."""
return RWLockFair._aWriter(self)
class RWLockReadD(RWLockableD):
"""A Read/Write lock giving preference to Reader."""
def __init__(self, lock_factory: Callable[[], Lockable] = threading.Lock, time_source: Callable[[], float] = time.perf_counter) -> None:
"""Init."""
self.v_read_count: _ThreadSafeInt = _ThreadSafeInt(initial_value=0, lock_factory=lock_factory)
self.c_time_source = time_source
self.c_resource = lock_factory()
self.c_lock_read_count = lock_factory()
super().__init__()
class _aReader(Lockable):
def __init__(self, p_RWLock: "RWLockReadD") -> None:
self.c_rw_lock = p_RWLock
self.v_locked: bool = False
super().__init__()
self.c_rw_lock.c_resource.release()
def locked(self) -> bool:
"""Answer to 'is it currently locked?'."""
return self.v_locked
def gen_rlock(self) -> "RWLockReadD._aReader":
"""Generate a reader lock."""
return RWLockReadD._aReader(self)
def gen_wlock(self) -> "RWLockReadD._aWriter":
"""Generate a writer lock."""
return RWLockReadD._aWriter(self)
class RWLockWriteD(RWLockableD):
"""A Read/Write lock giving preference to Writer."""
def __init__(self, lock_factory: Callable[[], Lockable] = threading.Lock, time_source: Callable[[], float] = time.perf_counter) -> None:
"""Init."""
self.v_read_count: _ThreadSafeInt = _ThreadSafeInt(lock_factory=lock_factory, initial_value=0)
self.v_write_count: int = 0
self.c_time_source = time_source
self.c_lock_read_count = lock_factory()
self.c_lock_write_count = lock_factory()
self.c_lock_read_entry = lock_factory()
self.c_lock_read_try = lock_factory()
self.c_resource = lock_factory()
super().__init__()
class _aReader(Lockable):
def __init__(self, p_RWLock: "RWLockWriteD") -> None:
self.c_rw_lock.c_lock_write_count.release()
def locked(self) -> bool:
"""Answer to 'is it currently locked?'."""
return self.v_locked
def gen_rlock(self) -> "RWLockWriteD._aReader":
"""Generate a reader lock."""
return RWLockWriteD._aReader(self)
def gen_wlock(self) -> "RWLockWriteD._aWriter":
"""Generate a writer lock."""
return RWLockWriteD._aWriter(self)
class RWLockFairD(RWLockableD):
"""A Read/Write lock giving fairness to both Reader and Writer."""
def __init__(self, lock_factory: Callable[[], Lockable] = threading.Lock, time_source: Callable[[], float] = time.perf_counter) -> None:
"""Init."""
self.v_read_count: int = 0
self.c_time_source = time_source
self.c_lock_read_count = lock_factory()
self.c_lock_read = lock_factory()
self.c_lock_write = lock_factory()
super().__init__()
class _aReader(Lockable):
def __init__(self, p_RWLock: "RWLockFairD") -> None:
self.c_rw_lock = p_RWLock
self.v_locked: bool = False
super().__init__()