diff --git a/tests/test_single_instance.py b/tests/test_single_instance.py index c87980b..1543a16 100644 --- a/tests/test_single_instance.py +++ b/tests/test_single_instance.py @@ -1,96 +1,71 @@ import pytest -from unittest.mock import patch +import sys +from unittest.mock import MagicMock, patch -import common -import wecker +# Mock hardware modules before importing wecker (tests run off the Pi). +sys.modules["RPi"] = MagicMock() +sys.modules["RPi.GPIO"] = MagicMock() +sys.modules["pygame"] = MagicMock() + +import common # noqa: E402 +import wecker # noqa: E402 -# We need to mock the PID file existence and os.kill to test the logic -# because we are running in a test environment. @pytest.fixture def mock_pid_file(tmp_path): - """Provides a temporary PID file path and ensures it's cleaned up.""" + """Provides a temporary PID file path and resets the lock fd.""" pid_file = tmp_path / "wecker.pid" with patch("wecker.PID_FILE", str(pid_file)): + wecker._pid_lock_fd = None yield pid_file + if wecker._pid_lock_fd is not None: + try: + wecker._pid_lock_fd.close() + except OSError: + pass + wecker._pid_lock_fd = None + def test_ensure_single_instance_success(mock_pid_file): - """Test that the script can start if no PID file exists.""" - # Ensure file doesn't exist - if mock_pid_file.exists(): - mock_pid_file.unlink() - - with patch("os.path.exists", return_value=False), \ - patch("os.getpid", return_value=1234): - + """Test that the script can start if no other instance holds the lock.""" + with patch("os.getpid", return_value=1234): wecker.ensure_single_instance() - - # Verify it wrote the PID - assert mock_pid_file.exists() - assert mock_pid_file.read_text() == "1234" + + assert mock_pid_file.read_text() == "1234" + def test_ensure_single_instance_already_running(mock_pid_file): - """Test that the script exits if another instance is running.""" - # 1. Create the PID file with a dummy PID - mock_pid_file.write_text("5678") - - # 2. Mock os.path.exists to find the file - # 3. Mock os.kill to succeed (meaning process 5678 is alive) - # Note: We DON'T mock builtins.open here, so it reads the real file we just wrote. - with patch("os.path.exists", return_value=True), \ - patch("os.kill") as mock_kill: - - mock_kill.return_value = None # Success means process is alive - - with pytest.raises(SystemExit) as excinfo: - wecker.ensure_single_instance() - - assert excinfo.value.code == 1 - mock_kill.assert_called_with(5678, 0) - -def test_ensure_single_instance_stale_pid(mock_pid_file): - """Test that the script continues if the PID in the file is dead.""" - mock_pid_file.write_text("5678") - - with patch("os.path.exists", return_value=True), \ - patch("os.kill", side_effect=ProcessLookupError), \ - patch("os.getpid", return_value=1234): - + """Test that the script exits if another instance holds the lock.""" + with patch("fcntl.flock", side_effect=BlockingIOError), pytest.raises( + SystemExit + ) as excinfo: wecker.ensure_single_instance() - - # Should have overwritten with new PID - assert mock_pid_file.read_text() == "1234" + + assert excinfo.value.code == 1 + + +def test_ensure_single_instance_overwrites_stale_pid(mock_pid_file): + """Test that a stale PID file is overwritten once the lock is acquired.""" + mock_pid_file.write_text("5678") + + with patch("os.getpid", return_value=1234): + wecker.ensure_single_instance() + + assert mock_pid_file.read_text() == "1234" + def test_remove_pid_file_success(mock_pid_file): - """Test that remove_pid_file removes the correct PID file.""" - mock_pid_file.write_text("1234") - - with patch("os.path.exists", return_value=True), \ - patch("os.getpid", return_value=1234): - - wecker.remove_pid_file() - assert not mock_pid_file.exists() + """Test that remove_pid_file closes the lock and removes the file.""" + with patch("os.getpid", return_value=1234): + wecker.ensure_single_instance() -def test_remove_pid_file_wrong_pid(mock_pid_file): - """Test that remove_pid_file does NOT remove if PID doesn't match.""" - mock_pid_file.write_text("5678") - - with patch("os.path.exists", return_value=True), \ - patch("os.getpid", return_value=1234): - - wecker.remove_pid_file() - assert mock_pid_file.exists() + wecker.remove_pid_file() + assert not mock_pid_file.exists() def test_pid_file_defined_once_across_modules(): - """ - DRY principle: PID_FILE must be defined in common.py and imported by - both wecker.py and api/schema.py — not redefined in each. - """ - # wecker.py imports PID_FILE from common — verify it's the same object + """DRY principle: PID_FILE must be defined in common.py and imported.""" assert wecker.PID_FILE is common.PID_FILE, ( "wecker.PID_FILE must reference common.PID_FILE, not redefine it" ) - - # Verify the path ends with "wecker.pid" assert common.PID_FILE.endswith("wecker.pid") diff --git a/wecker.py b/wecker.py index 38deeb4..21c4c74 100644 --- a/wecker.py +++ b/wecker.py @@ -5,6 +5,7 @@ import pygame import random import sys import os +import fcntl from common import PID_FILE @@ -16,44 +17,42 @@ logging.basicConfig( ) +_pid_lock_fd = None + + def ensure_single_instance(): - """Ensures that only one instance of the script is running.""" - if os.path.exists(PID_FILE): - try: - with open(PID_FILE, 'r') as f: - old_pid = int(f.read().strip()) + """Ensures that only one instance of the script is running. - # Check if the process is still running - os.kill(old_pid, 0) - # If os.kill succeeds, the process is running. - logging.error(f"Another instance of wecker (PID {old_pid}) is already running. Exiting.") - sys.exit(1) - except (ValueError, ProcessLookupError): - # ValueError: PID file is not an integer - # ProcessLookupError: PID doesn't exist - pass - except PermissionError: - # Process exists but we don't have permission to signal it. - logging.error(f"Another instance of wecker (PID {old_pid}) is already running. Exiting.") - sys.exit(1) - except OSError: - # Other OS errors, might mean process is dead - pass + Uses an advisory file lock on the PID file so the check-and-write + sequence is atomic and race-free. + """ + global _pid_lock_fd + _pid_lock_fd = open(PID_FILE, "a+") + try: + fcntl.flock(_pid_lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + except (BlockingIOError, OSError): + logging.error("Another instance of wecker is already running. Exiting.") + sys.exit(1) - with open(PID_FILE, 'w') as f: - f.write(str(os.getpid())) + _pid_lock_fd.seek(0) + _pid_lock_fd.truncate() + _pid_lock_fd.write(str(os.getpid())) + _pid_lock_fd.flush() def remove_pid_file(): - """Removes the PID file.""" - if os.path.exists(PID_FILE): + """Releases the PID file lock and removes the file.""" + global _pid_lock_fd + if _pid_lock_fd is not None: try: - with open(PID_FILE, 'r') as f: - current_pid = int(f.read().strip()) - if current_pid == os.getpid(): - os.remove(PID_FILE) - except (ValueError, OSError): + _pid_lock_fd.close() + except OSError: pass + _pid_lock_fd = None + try: + os.remove(PID_FILE) + except FileNotFoundError: + pass # GPIO Setup