From a744fbb435ef3c2b229d7ba5ad4b824f1577f6db Mon Sep 17 00:00:00 2001 From: Christian Berendt Date: Mon, 3 Aug 2026 10:12:11 +0200 Subject: [PATCH] Add integration tests for the Redis semaphore RedisSemaphore keeps its holders in a sorted set and admits them from a server-side Lua script. The unit suite already runs that script against fakeredis, which executes Lua through lupa, so its logic is covered there. What fakeredis cannot cover is Redis' own Lua sandbox and redis.call bindings, the numeric coercion of the now, maxsize and expiry arguments that lupa only approximates, and a race between many clients for the same free slot, which a single sequential client cannot stage. The new module runs three tests against the live Redis of the python-osism-integration-tests job: a slot freed by release being granted to a caller that was refused inside its retry loop, a roundtrip through create_netbox_semaphore including the key it derives from a NetBox URL, and the many-client race, where twelve barrier-synchronised threads contend for three slots and exactly three get in. The race runs five rounds, because a single round lands the interleaving that exposes an over-admitting implementation about nine times in ten. Its acquire timeout is kept short, since the losers of a round wait theirs out and that is what the round costs. Every thread returns its result through a future, so a worker that dies fails the test instead of quietly reducing the number of contenders. Each test uses a uuid key and deletes it afterwards, so runs stay independent of one another. No production code and no dependency changes. Assisted-by: Claude:claude-fable-5 Assisted-by: Claude:claude-opus-5[1m] Signed-off-by: Christian Berendt --- tests/integration/test_semaphore.py | 149 ++++++++++++++++++++++++++++ 1 file changed, 149 insertions(+) create mode 100644 tests/integration/test_semaphore.py diff --git a/tests/integration/test_semaphore.py b/tests/integration/test_semaphore.py new file mode 100644 index 000000000..7a55b8313 --- /dev/null +++ b/tests/integration/test_semaphore.py @@ -0,0 +1,149 @@ +# SPDX-License-Identifier: Apache-2.0 + +"""Redis-semaphore integration tests against a live Redis. + +``RedisSemaphore`` caps concurrent NetBox API requests: ``osism.tasks.netbox`` +builds one per NetBox URL through ``create_netbox_semaphore``. Holders live in a +sorted set and are admitted by a server-side Lua script (``ZREMRANGEBYSCORE`` +plus ``ZCARD`` plus ``ZADD``). + +The script's logic is not what needs a live server: +``tests/unit/utils/test_init_semaphore.py`` runs the production ``_ACQUIRE_LUA`` +against ``fakeredis``, which executes Lua through ``lupa``. What only the real +server provides is Redis' own Lua sandbox and ``redis.call`` bindings, the +numeric coercion of the ``now``, ``maxsize`` and expiry arguments that ``lupa`` +only approximates, and the race the unit suite explicitly defers to here: +many clients contending for the same free slot at the same moment. +""" + +import concurrent.futures +import hashlib +import threading +import uuid + +import pytest + +from osism import utils + +pytestmark = pytest.mark.integration + + +def semaphore_redis_key(key): + """The prefixed key ``RedisSemaphore`` stores its sorted set under.""" + return f"semaphore:{key}" + + +@pytest.fixture +def redis_client(): + """The shared Redis client the semaphore itself uses. + + ``create_netbox_semaphore`` wires this very client into the semaphores it + builds, so seeding and inspecting through it observes what the code under + test sees. It is a cached module global held for the whole session, so it + is not closed here: that would break every later test that touches it. + """ + return utils.redis + + +@pytest.fixture +def semaphore_key(redis_client): + """A key unique to this test, removed from Redis afterwards. + + ``RedisSemaphore`` prefixes what it is handed, so the sorted set lives under + ``semaphore:``. Raw seeding and inspection use that prefixed form. + """ + key = f"itest-sem-{uuid.uuid4()}" + yield key + redis_client.delete(semaphore_redis_key(key)) + + +def test_release_frees_slot(redis_client, semaphore_key): + """The slot a holder gives up is granted to the caller that was refused. + + The unit suite covers a slot freed by expiry, never one freed by + ``release`` and granted to a waiter inside its retry loop. + """ + holder = utils.RedisSemaphore(redis_client, semaphore_key, 1, timeout=5) + assert holder.acquire() is True + + waiter = utils.RedisSemaphore(redis_client, semaphore_key, 1, timeout=0.5) + assert waiter.acquire() is False + + holder.release() + + assert waiter.acquire() is True + assert redis_client.zcard(waiter.key) == 1 + + waiter.release() + + +def test_create_netbox_semaphore_key_and_maxsize(redis_client): + """The NetBox helper derives its key from the URL and works against Redis.""" + url = f"https://netbox-{uuid.uuid4()}.example" + url_hash = hashlib.md5(url.encode()).hexdigest()[:8] + redis_key = f"semaphore:netbox_semaphore_{url_hash}" + + sem = utils.create_netbox_semaphore(url, max_connections=2) + + try: + assert isinstance(sem, utils.RedisSemaphore) + assert sem.key == redis_key + assert sem.maxsize == 2 + + assert sem.acquire() is True + members = {member.decode() for member in redis_client.zrange(redis_key, 0, -1)} + assert members == {sem.identifier} + + sem.release() + assert redis_client.zcard(redis_key) == 0 + finally: + redis_client.delete(redis_key) + + +def test_concurrent_acquire_never_exceeds_maxsize(redis_client, semaphore_key): + """Under a real race only ``maxsize`` of many contending clients get in. + + The atomic Lua script is what makes this hold: a capacity check and a slot + reservation split across two round trips would let several clients observe + the same free slot and all take it. This is the many-client race the unit + suite defers to a live server; ``fakeredis`` never reproduces it, because a + single sequential client cannot over-admit. + + One round only lands the interleaving that exposes an over-admitting + implementation about nine times in ten, so the race is run repeatedly. A + round costs a full acquire timeout, since the losers each wait theirs out, + which is why the timeout here is much shorter than elsewhere in this file. + + Every thread hands its result back through a future, so a worker that dies + fails the test. Left unchecked it would only warn, and the race this test + exists for would silently shrink to the threads that survived. + """ + maxsize = 3 + thread_count = 12 + rounds = 5 + redis_key = semaphore_redis_key(semaphore_key) + + def contend(barrier): + sem = utils.RedisSemaphore(redis_client, semaphore_key, maxsize, timeout=0.1) + barrier.wait() + return sem if sem.acquire() else None + + for _ in range(rounds): + # A timeout on the barrier turns a thread that never arrives into a + # failed assertion rather than a hung CI job. + barrier = threading.Barrier(thread_count, timeout=30) + + with concurrent.futures.ThreadPoolExecutor(max_workers=thread_count) as pool: + futures = [pool.submit(contend, barrier) for _ in range(thread_count)] + acquired = [sem for sem in (future.result() for future in futures) if sem] + + # Nothing is released before every future has been collected, so the + # holder count is stable here. + assert len(acquired) == maxsize + assert redis_client.zcard(redis_key) == maxsize + + for sem in acquired: + sem.release() + + assert redis_client.zcard(redis_key) == 0 + redis_client.delete(redis_key)