Part of #2400. Follow-up to the integration-test foundation (#2368).
What
Cover osism.utils.RedisSemaphore and the create_netbox_semaphore helper (osism/utils/__init__.py:109 and :472) against the live Redis from the integration job.
The semaphore is implemented on top of Redis sorted sets (zadd / zcard / zremrangebyscore). This is exactly the kind of behaviour a fakeredis-based unit test can get wrong, so it belongs in the integration suite.
Scope
- Acquire up to
maxsize holders successfully.
- An additional
acquire(timeout=...) beyond maxsize returns False after the timeout.
release() frees a slot so a subsequent acquire() succeeds.- Expired holders (score older than 60s) are cleaned up by
zremrangebyscore and free a slot — seed an aged entry directly via the Redis client to exercise this deterministically.
- Context-manager usage (
with sem:) acquires and releases; TimeoutError is raised when it cannot acquire.
create_netbox_semaphore(url, max_connections=...) builds a semaphore on the expected semaphore:netbox:<url> key with the expected maxsize.
Where
New file tests/integration/test_semaphore.py, pytestmark = pytest.mark.integration. Use unique keys (uuid) per test to stay independent of prior runs.
Part of #2400. Follow-up to the integration-test foundation (#2368).
What
Cover
osism.utils.RedisSemaphoreand thecreate_netbox_semaphorehelper (osism/utils/__init__.py:109and:472) against the live Redis from the integration job.The semaphore is implemented on top of Redis sorted sets (
zadd/zcard/zremrangebyscore). This is exactly the kind of behaviour afakeredis-based unit test can get wrong, so it belongs in the integration suite.Scope
maxsizeholders successfully.acquire(timeout=...)beyondmaxsizereturnsFalseafter the timeout.release()frees a slot so a subsequentacquire()succeeds.zremrangebyscoreand free a slot — seed an aged entry directly via the Redis client to exercise this deterministically.with sem:) acquires and releases;TimeoutErroris raised when it cannot acquire.create_netbox_semaphore(url, max_connections=...)builds a semaphore on the expectedsemaphore:netbox:<url>key with the expectedmaxsize.Where
New file
tests/integration/test_semaphore.py,pytestmark = pytest.mark.integration. Use unique keys (uuid) per test to stay independent of prior runs.