Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def setUp(self) -> None:
"""Test setup."""
self.c_rwlock_type_downgradable = (rwlock.RWLockWriteD, rwlock.RWLockFairD, rwlock.RWLockReadD)
self.c_rwlock_type = (rwlock.RWLockRead, rwlock.RWLockWrite, rwlock.RWLockFair) + self.c_rwlock_type_downgradable
def test_read_vs_downgrade_write(self) -> None:
"""
# Given: Instance of RWLockWriteD.
# When: A reader lock is acquired OR A writer lock is downgraded.
# Then: The internal state should be the same.
"""
# ## Arrange
c_rwlock_1 = rwlock.RWLockWriteD()
c_rwlock_2 = rwlock.RWLockWriteD()
def assert_internal_state():
self.assertEqual(int(c_rwlock_1.v_read_count), int(c_rwlock_2.v_read_count))
self.assertEqual(int(c_rwlock_1.v_write_count), int(c_rwlock_2.v_write_count))
self.assertEqual(bool(c_rwlock_1.c_lock_read_count.locked()), bool(c_rwlock_2.c_lock_read_count.locked()))
self.assertEqual(bool(c_rwlock_1.c_lock_write_count.locked()), bool(c_rwlock_2.c_lock_write_count.locked()))
self.assertEqual(bool(c_rwlock_1.c_lock_read_entry.locked()), bool(c_rwlock_2.c_lock_read_entry.locked()))
self.assertEqual(bool(c_rwlock_1.c_lock_read_try.locked()), bool(c_rwlock_2.c_lock_read_try.locked()))
self.assertEqual(bool(c_rwlock_1.c_resource.locked()), bool(c_rwlock_2.c_resource.locked()))
# ## Assume
assert_internal_state()
# ## Act
def test_read_vs_downgrade_write(self) -> None:
"""
# Given: Instance of RWLockWriteD.
# When: A reader lock is acquired OR A writer lock is downgraded.
# Then: The internal state should be the same.
"""
# ## Arrange
c_rwlock_1 = rwlock.RWLockWriteD()
c_rwlock_2 = rwlock.RWLockWriteD()
def assert_internal_state():
self.assertEqual(int(c_rwlock_1.v_read_count), int(c_rwlock_2.v_read_count))
self.assertEqual(int(c_rwlock_1.v_write_count), int(c_rwlock_2.v_write_count))
self.assertEqual(bool(c_rwlock_1.c_lock_read_count.locked()), bool(c_rwlock_2.c_lock_read_count.locked()))
self.assertEqual(bool(c_rwlock_1.c_lock_write_count.locked()), bool(c_rwlock_2.c_lock_write_count.locked()))
self.assertEqual(bool(c_rwlock_1.c_lock_read_entry.locked()), bool(c_rwlock_2.c_lock_read_entry.locked()))
self.assertEqual(bool(c_rwlock_1.c_lock_read_try.locked()), bool(c_rwlock_2.c_lock_read_try.locked()))
self.assertEqual(bool(c_rwlock_1.c_resource.locked()), bool(c_rwlock_2.c_resource.locked()))
# ## Assume
assert_internal_state()
# ## Act
a_read_lock = c_rwlock_1.gen_rlock()
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)