(self)
| 902 | self.semaphore.acquire('a', blocking=False) |
| 903 | |
| 904 | def test_release_capacity(self): |
| 905 | acquire_token = self.semaphore.acquire('a', blocking=False) |
| 906 | self.semaphore.release('a', acquire_token) |
| 907 | try: |
| 908 | self.semaphore.acquire('a', blocking=False) |
| 909 | except NoResourcesAvailable: |
| 910 | self.fail( |
| 911 | 'The release of the semaphore should have allowed for ' |
| 912 | 'the second acquire to not be blocked' |
| 913 | ) |
| 914 | |
| 915 | |
| 916 | class TestSlidingWindowSemaphore(unittest.TestCase): |