源码: tests/bounded_semaphore.py
# -.- coding:utf-8 -.-
import unittest
import threading
class TestBoundedSemaphore(unittest.TestCase):
"""
Bounded Semaphore 是基于 Semaphore 基础上的上限限制对象.
"""
def test_acquire_multi_times(self):
# acquire 行为无变化, 因为Bounded Semaphore
# 沿用的是Semaphore的acquire方法.
sem = threading.BoundedSemaphore(2)
status = sem.acquire()
self.assertTrue(status)
status = sem.acquire()
self.assertTrue(status)
status = sem.acquire(blocking=0)
self.assertFalse(status)
def test_release_unacquire(self):
# BoundedSemaphore实例化的上限参数是2,
# sem.release() 想要给 sem 的上限 +1,
# 由于BoundedSemaphore重新实现了release()方法,
# 它判断了, 如果 现有值已经是上限值, 那么就报错!
# 也就是说, 不允许再上限值的基础上+1.
sem = threading.BoundedSemaphore(2)
self.assertRaises(ValueError, sem.release)
测试: tests/main.py
import unittest
TEST_MODULE = [
"ln_threading.tests.bounded_semaphore",
]
if __name__ == '__main__':
suite = unittest.defaultTestLoader.loadTestsFromNames(TEST_MODULE)
runner = unittest.TextTestRunner(verbosity=2)
runner.run(suite)
