65 lines
2 KiB
Python
65 lines
2 KiB
Python
|
"""Unit tests for the single-instance leader election (singleton.py)."""
|
||
|
|
import fcntl
|
||
|
|
import importlib
|
||
|
|
import os
|
||
|
|
|
||
|
|
import singleton
|
||
|
|
|
||
|
|
|
||
|
|
def _fresh():
|
||
|
|
# _held is module-level state; reload for an isolated starting point per test.
|
||
|
|
return importlib.reload(singleton)
|
||
|
|
|
||
|
|
|
||
|
|
def test_no_lock_path_is_always_leader():
|
||
|
|
s = _fresh()
|
||
|
|
assert s.claim(None) is True
|
||
|
|
assert s.claim("") is True
|
||
|
|
|
||
|
|
|
||
|
|
def test_first_claimer_wins_and_holds_the_lock(tmp_path):
|
||
|
|
s = _fresh()
|
||
|
|
lock = str(tmp_path / "notifier.lock")
|
||
|
|
assert s.claim(lock) is True
|
||
|
|
# Held for the process lifetime: the fd stays open and flock'd.
|
||
|
|
assert lock in s._held
|
||
|
|
|
||
|
|
|
||
|
|
def test_reclaiming_same_path_is_idempotent(tmp_path):
|
||
|
|
s = _fresh()
|
||
|
|
lock = str(tmp_path / "notifier.lock")
|
||
|
|
assert s.claim(lock) is True
|
||
|
|
fd = s._held[lock]
|
||
|
|
# Re-claiming must not re-open or drop the lock — same fd, still True.
|
||
|
|
assert s.claim(lock) is True
|
||
|
|
assert s._held[lock] == fd
|
||
|
|
|
||
|
|
|
||
|
|
def test_second_holder_stands_down(tmp_path):
|
||
|
|
"""When another holder already owns the lock, claim() returns False and leaks no
|
||
|
|
fd. Simulate the other worker by taking the exclusive flock on a separate open
|
||
|
|
file description (a fresh os.open, as a distinct process would)."""
|
||
|
|
s = _fresh()
|
||
|
|
lock = str(tmp_path / "notifier.lock")
|
||
|
|
other = os.open(lock, os.O_RDWR | os.O_CREAT, 0o644)
|
||
|
|
fcntl.flock(other, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||
|
|
try:
|
||
|
|
assert s.claim(lock) is False
|
||
|
|
assert lock not in s._held # did not record a role it doesn't own
|
||
|
|
finally:
|
||
|
|
fcntl.flock(other, fcntl.LOCK_UN)
|
||
|
|
os.close(other)
|
||
|
|
|
||
|
|
|
||
|
|
def test_reelection_after_holder_releases(tmp_path):
|
||
|
|
"""A restart drops the OS lock; the next claim() then wins."""
|
||
|
|
s = _fresh()
|
||
|
|
lock = str(tmp_path / "notifier.lock")
|
||
|
|
other = os.open(lock, os.O_RDWR | os.O_CREAT, 0o644)
|
||
|
|
fcntl.flock(other, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||
|
|
assert s.claim(lock) is False
|
||
|
|
# Prior leader exits -> OS releases the lock.
|
||
|
|
fcntl.flock(other, fcntl.LOCK_UN)
|
||
|
|
os.close(other)
|
||
|
|
assert s.claim(lock) is True
|