Unit tests for osism/utils/__init__.py — semaphore, redlock, task locks

[berendt]

Background

Follow-up to #2192 (foundation) and PR #2193 (pytest + Zuul infrastructure). Part of Tier 4 (#2199). Companion to the connection-init and task-output issues: covers the concurrency primitives (RedisSemaphore, redlock, NetBox semaphore) and the global task-lock helpers from osism/utils/__init__.py.

Scope

Add tests/unit/utils/test_init_locks.py covering the helpers below in osism/utils/__init__.py.

Test targets

RedisSemaphore — __init__.py:109

Use a MagicMock for the redis client. Drive time.time() and time.sleep() either via freezegun or mocker.patch so the loop exits deterministically.

__init__

• Stores redis, key=f"semaphore:{key}", maxsize, timeout; identifier=None
• key prefixing is applied even when the input already starts with semaphore: (no double-strip — confirm production behaviour)

acquire(timeout=None)

• zcard < maxsize on first try → zadd called with {identifier: now}, self.identifier set, returns True
• zcard >= maxsize for entire window → returns False
• zremrangebyscore(self.key, 0, now-60) called every iteration (cleanup)
• Default timeout fallback: acquire(None) with self.timeout=None → uses 10 seconds
• Explicit acquire(timeout=5) → overrides instance timeout
• Identifier is a fresh uuid.uuid4() per call (verify by patching uuid.uuid4)
• time.sleep(0.01) called between retries when full

release()

• Sets identifier → zrem(self.key, self.identifier) called and identifier=None
• Called twice → second call is a no-op (no extra zrem)
• Called without acquire → no-op

Context manager (__enter__ / __exit__)

• acquire returns True → __enter__ returns self
• acquire returns False → __enter__ raises TimeoutError containing the key
• __exit__ always calls release() and returns False (don't swallow exceptions)

create_redlock(key, auto_release_time=3600) — __init__.py:441

Patch pottery.Redlock, _init_redis, logging.getLogger.

• Returns the Redlock instance built with key, masters={redis}, auto_release_time
• Default auto_release_time=3600
• Custom auto_release_time=600 propagated
• Pottery logger is set to CRITICAL level (verify setLevel(logging.CRITICAL) call on the "pottery" logger)
• stdout/stderr suppressed during construction (the production code uses redirect_stdout(devnull)/redirect_stderr(devnull) — assert no construction-time noise leaks; one assertion via capsys is enough)

create_netbox_semaphore(netbox_url, max_connections=None) — __init__.py:469

Patch _init_redis, osism.utils.settings.NETBOX_MAX_CONNECTIONS.

• max_connections=None → uses settings.NETBOX_MAX_CONNECTIONS
• Explicit max_connections=20 → propagated
• Returns a RedisSemaphore with key=f"netbox_semaphore_{md5_hash[:8]}", timeout=30, redis_client=_init_redis()
• Two different URLs → two different keys (verify the hash is per-URL)
• Same URL twice → identical key

set_task_lock(user=None, reason=None) — __init__.py:497

Patch _init_redis, osism.utils.settings.OPERATOR_USER. Capture the JSON written via redis.set.

• user=None → falls back to settings.OPERATOR_USER
• Explicit user="alice" → used directly
• reason=None → stored as null in JSON
• Captured JSON contains locked=True, ISO-formatted timestamp, user, reason
• redis.set raises → returns False, error logged

remove_task_lock() — __init__.py:526

• Calls delete("osism:task_lock") → returns True
• delete raises → returns False, error logged

is_task_locked() — __init__.py:541

• Redis returns None → returns None
• Redis returns valid JSON bytes → returns the parsed dict (verify .decode("utf-8") is applied)
• redis.get raises → returns None, error logged
• JSON-decode error → returns None, error logged

check_task_lock_and_exit() — __init__.py:641

Patch is_task_locked and builtins.exit.

• No lock → no exit, returns None
• Locked, all fields present → logs the user/timestamp/reason, then calls exit(1)
• Locked, reason=None → Reason: line not logged
• Locked, missing user/timestamp keys → defaults "unknown" used in log

Mocking hints

• For RedisSemaphore, patch time.time to return a controlled sequence so the while time.time() < end_time loop terminates predictably:

  mocker.patch("osism.utils.time.time", side_effect=[0, 0.001, 11])
  mocker.patch("osism.utils.time.sleep")

• For create_redlock, the production code does from pottery import Redlock inside the function, so patch pottery.Redlock.
• set_task_lock JSON payload includes a datetime.now().isoformat() value — match it with a regex (\d{4}-\d{2}-\d{2}T...) instead of an exact string, or freeze time.
• The pottery logger setting is mutable global state — verify the call but reset the level in a fixture if you care about test isolation.

Definition of Done

• tests/unit/utils/test_init_locks.py created
• All listed cases covered
• pytest --cov=osism.utils for the targeted helpers ≥ 90 %
• pipenv run pytest tests/unit/utils/test_init_locks.py passes locally
• flake8, mypy, python-black remain green
• Zuul job python-osism-unit-tests passes

Dependencies

• Part of Meta: Unit test coverage for osism/ #2199 (meta). Blocked by Introduce foundation for unit tests (pytest + Zuul CI integration) #2192 / PR Introduce foundation for unit tests (pytest + Zuul CI) #2193.