thermograph/tests/core/test_singleton.py

"""Unit tests for the single-instance leader election (singleton.py)."""
import fcntl
import importlib
import os

from core 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