Add integration tests for the Redis semaphore

[berendt]

RedisSemaphore caps concurrent NetBox API requests: osism/tasks/netbox.py:31 builds one per NetBox URL through utils.create_netbox_semaphore(nb.base_url). Its holders live in a sorted set and are admitted by a server-side Lua script, so what the class does is Redis behaviour: the expiry reclaim, the capacity limit, and the atomicity that stops two clients from taking the same free slot. Until now the only coverage came from fakeredis and mocks, and fakeredis reimplements exactly that surface in Python.

One commit, one new file, no production code and no dependency changes.

Add integration tests for the Redis semaphore adds tests/integration/test_semaphore.py with ten tests against the live Redis of the python-osism-integration-tests job:

• test_acquire_up_to_maxsize, test_acquire_beyond_maxsize_returns_false, test_release_frees_slot and test_release_without_acquire_is_noop cover the capacity limit and both release paths.
• test_expired_holder_reclaimed seeds a holder aged past HOLDER_EXPIRY directly through the Redis client, so the ZREMRANGEBYSCORE reclaim runs without waiting out 60 seconds.
• test_context_manager_roundtrip and test_context_manager_timeout_raises cover both with sem: paths, including the exact TimeoutError message.
• test_create_netbox_semaphore_key_and_maxsize and test_create_netbox_semaphore_default_maxsize pin the key the helper derives from a NetBox URL (semaphore:netbox_semaphore_<md5-prefix>), its maxsize, and an acquire/release round trip against the live server.
• test_concurrent_acquire_never_exceeds_maxsize is the many-client race the unit suite defers to a real server (tests/unit/utils/test_init_semaphore.py:8): twelve barrier-synchronised threads contend for three slots and exactly three get in.

Each test uses a uuid key and deletes it afterwards, so runs stay independent of one another.

Verification

Against a throwaway redis:7-alpine container, pytest tests/integration/test_semaphore.py reports 10 passed, repeated three times without flakiness. With Redis stopped it reports 10 skipped rather than failing, per the skip logic in tests/integration/conftest.py.

The race test was checked for teeth: replaying the same twelve threads with a deliberately non-atomic acquire (ZCARD then ZADD in two round trips) admitted 6 to 7 holders where maxsize is 3, so the test fails loudly on the bug the Lua script prevents.

Closes #2401

[sourcery-ai]

Hey - I've reviewed your changes and they look great!

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}

[ideaship]

This rationale isn't accurate, and it's load-bearing — it's the stated reason the file exists. tests/unit/utils/test_init_semaphore.py:4-7 runs the production _ACQUIRE_LUA against fakeredis with lupa (Pipfile:51-52 declares both), so the unit suite already executes the real server-side script rather than reimplementing it in Python or asserting on a mock's recorded calls.

There is a genuine argument for a live server here, and it's narrower: Redis' own Lua sandbox, its redis.call bindings, and the numeric coercion of the now / maxsize / expiry arguments, which lupa only approximates — plus the many-client race that the unit docstring itself carves out at :8 as needing "a real Redis". Worth stating that instead. The commit message carries the same claim and would want the same correction.

[ideaship]

The "semaphore:" prefix is rebuilt by hand at eleven places in this file (:47, 60, 65, 77, 93, 105, 114, 130, 143, 162, 198) while sem.key — which is public — is used only at :168. A prefix change would break eleven assertions for reasons unrelated to what each test is checking.

Not all eleven want the same treatment. Where a semaphore is already in scope (:60, 65, 77, 93, 105) sem.key substitutes directly. :114, 130, 143 need only a one-line reorder to construct the semaphore first. For this fixture and the per-thread constructor at :198 there's no instance yet, so a small helper is the right shape:

def semaphore_redis_key(key):
    return f"semaphore:{key}"

Keep :162 hand-written — that one is the assertion in assert sem.key == redis_key, and substituting sem.key there would delete the test.

[ideaship]

Scope question for the file as a whole, and it splits three ways rather than two — the overlap with the unit suite isn't all the same kind.

Genuinely redundant — delete. test_acquire_up_to_maxsize (:50) and test_acquire_beyond_maxsize_returns_false (:68) are tests/unit/utils/test_init_semaphore.py:43 test_acquire_never_exceeds_maxsize split in two, and the unit version asserts strictly more (it also pins that the refused caller leaves the set unchanged). test_expired_holder_reclaimed (:108) is weaker than its unit twin at :79: seeding at time.time() - HOLDER_EXPIRY - 1 (:115) means the holder is already expired, so the first eval reclaims it and the retry loop never runs, whereas test_reclaim_boundary_advances_across_retries seeds at "now" with a shrunk HOLDER_EXPIRY so the holder ages out mid-loop. Mutation check: hoist now = time.time() out of the while in acquire() (osism/utils/__init__.py:173) — the unit test goes red, this one stays green.

Not redundant, but no live-Redis signal — move to tests/unit/, don't delete. No unit test currently covers release() short-circuiting on identifier is None (:98), __enter__/__exit__ (:128), the TimeoutError raise from __enter__ (:141), or the max_connections fallback (:181). Deleting these loses real coverage; they're simply at the wrong layer. :98 issues no Lua at all — release() returns before zrem, so the only server contact is a zcard on a key nobody wrote — :128 exercises four lines of pure Python, and :181 performs no semaphore operation whatsoever.

Keep. test_release_frees_slot (:82) is not a duplicate: the unit suite covers a slot freed by expiry, never one freed by release and granted to a waiter mid-retry-loop. test_create_netbox_semaphore_key_and_maxsize (:158) is a real roundtrip through the production helper. test_concurrent_acquire_never_exceeds_maxsize (:188) is the file's reason to exist.

That leaves the Redis-gated job running what actually needs Redis, and drops ~1.2 s of real time.sleep retry loops (timeout=0.5, 0.5, 0.2) from it.

[ideaship]

This one performs no semaphore operation, and its assertion reads the same setting the helper reads (osism/utils/__init__.py:516-517), so it largely asserts that a value equals itself. It does still detect a literal substitution or a deleted fallback branch, so it's worth keeping rather than dropping — but as a unit test with a sentinel:

mocker.patch.object(settings, "NETBOX_MAX_CONNECTIONS", 17)
assert utils.create_netbox_semaphore("https://netbox.example").maxsize == 17

That pins the fallback branch instead of the tautology. Note it isn't free of Redis today: create_netbox_semaphore calls _init_redis() (osism/utils/__init__.py:528), which constructs a client and pings, so the test currently needs a reachable server to build an object it never uses. Moved to the unit suite it wants _init_redis mocked — tests/unit/utils/test_init_connections.py already has the idiom.

[ideaship]

This is the only test that exercises the invariant under real concurrency, and it runs the race exactly once. Worth being precise about what that covers, because the obvious reading is stronger than the truth.

A revert of the atomic script to separate round trips is caught deterministically elsewhere: tests/unit/utils/test_init_locks.py:61-83 and five neighbours assert redis.eval.assert_called_once_with(_ACQUIRE_LUA, …), so they fail whether or not an interleaving lands. And the race cannot be reintroduced inside the script, since Redis executes an EVAL atomically. So this test is not what stands between a straight revert and a release.

Where it is the only guard is a refactor rather than a revert — register_script/evalsha, say — where those mock assertions get rewritten to match the new call, and part of the capacity decision moves client-side in the process. The rewritten shape tests bless the new shape, the sequential fakeredis tests cannot over-admit with a single client, and this is the only test left that would notice.

For that case, one round is roughly a 1-in-10 miss. With acquire() mutated to three round trips, this test caught it in 54 of 60 cold-process runs against a live Redis (and in 0 of 100 runs against fakeredis, which is why it belongs here rather than in the unit suite). Wrapping the barrier, threads and assertions in for _ in range(5) with a redis_client.delete(redis_key) between rounds takes the miss probability to ~1e-5.

One cost note, since this file's runtime is already mostly waiting: a round costs a full acquire timeout, because the nine losers each wait theirs out — 0.51 s measured. Five rounds at timeout=0.5 would add ~2 s. Dropping this test's acquire timeout to 0.1 makes a round 0.11 s with detection unchanged (58 of 60 measured), so five rounds cost about what one round costs today.

[ideaship]

contend() has no error handling, and nothing propagates a worker failure back to the test thread. A thread that dies inside acquire() goes through threading.excepthook, pytest's threadexception plugin downgrades it to a PytestUnhandledThreadExceptionWarning, and setup.cfg:184 sets addopts = -ra --strict-markers with no filterwarnings = error — so the test still passes.

Concretely: three threads win their slots and the other nine die on a transient error after crossing the barrier but before contending. len(acquired) == 3 and zcard == 3 both hold, the job goes green, and the twelve-way race this file exists to exercise silently degenerated into a three-way one. That matters more here than the general unchecked-thread pattern, because this is the only behavioural check on the atomicity guarantee — and it compounds the single-round point above: the one round that carries the signal can quietly not happen.

Simplest fix is to drop the manual threads for a pool and call result() on every future, so propagation is automatic rather than opt-in. It also removes the acquired_lock bookkeeping:

with concurrent.futures.ThreadPoolExecutor(max_workers=thread_count) as pool:
    futures = [pool.submit(contend) for _ in range(thread_count)]
    acquired = [sem for sem in (f.result() for f in futures) if sem]

with contend() returning sem or None. Note the barrier keeps working unchanged: a thread that dies before barrier.wait() leaves the rest to time out after 30 s with BrokenBarrierError, which now surfaces as a failure instead of a warning.