From 558c4ff5b6f56c571c09a259f4c4558b04a278c1 Mon Sep 17 00:00:00 2001 From: Markus Graf Date: Tue, 19 May 2026 15:15:05 +0200 Subject: [PATCH] Add isRinging query to check alarm state via API Features: - New GraphQL query 'isRinging' returns true/false if wecker is active - Checks the wecker process via PID file (PID_FILE in common.py) DRY refactoring: - Extract PID_FILE into shared common.py module - Both wecker.py and api/schema.py import from common - DRY enforcement tests verify identity (is) not just equality Tests: - test_is_ringing_returns_false_when_not_running - test_is_ringing_returns_true_when_running - test_pid_file_defined_once_across_modules (DRY enforcement) - test_pid_file_same_shared_constant_in_api (DRY enforcement) - Cleaned up unused imports in test_single_instance.py Docs: - Updated README.md with isRinging query documentation --- README.md | 17 ++++++++++++++++ api/schema.py | 19 ++++++++++++++++++ common.py | 5 +++++ tests/test_api.py | 38 +++++++++++++++++++++++++++++++++++ tests/test_single_instance.py | 20 +++++++++++++++--- wecker.py | 4 ++-- 6 files changed, 98 insertions(+), 5 deletions(-) create mode 100644 common.py diff --git a/README.md b/README.md index b1c82e4..561bcee 100644 --- a/README.md +++ b/README.md @@ -57,6 +57,14 @@ Instead of manually editing your crontab, this project provides a simple GraphQL ### Example API Usage +**Check if alarm is currently ringing:** +```graphql +query { + isRinging +} +``` +Returns `true` if the wecker alarm clock is currently active (playing music), `false` otherwise. + **Get all alarms:** ```graphql query { @@ -92,6 +100,15 @@ mutation { ``` +### Example response for `isRinging`: +```json +{ + "data": { + "isRinging": true + } +} +``` + ## How the Puzzle Works 1. **Ringing:** The music plays in an endless loop. 2. **Start:** Press the arcade button once to start the puzzle. Wait 3 seconds. diff --git a/api/schema.py b/api/schema.py index e7c311d..5545012 100644 --- a/api/schema.py +++ b/api/schema.py @@ -4,6 +4,7 @@ import os import sys from pathlib import Path from api.crontab_manager import CrontabManager +from common import PID_FILE def get_manager(): @@ -11,6 +12,20 @@ def get_manager(): return CrontabManager(tabfile=tabfile) +def is_wecker_ringing() -> bool: + """Check if the wecker process is currently running via the PID file.""" + if not os.path.exists(PID_FILE): + return False + try: + pid: int + with open(PID_FILE) as f: + pid = int(f.read().strip()) + os.kill(pid, 0) + return True + except (ValueError, ProcessLookupError, PermissionError, OSError): + return False + + @strawberry.type class Alarm: id: str @@ -21,6 +36,10 @@ class Alarm: @strawberry.type class Query: + @strawberry.field + def is_ringing(self) -> bool: + return is_wecker_ringing() + @strawberry.field def get_alarms(self) -> List[Alarm]: manager = get_manager() diff --git a/common.py b/common.py new file mode 100644 index 0000000..2637cde --- /dev/null +++ b/common.py @@ -0,0 +1,5 @@ +from pathlib import Path + +# The PID file path — defined once here to avoid duplication. +# wecker.py writes this file when it starts; api/schema.py reads it to check state. +PID_FILE = str(Path(__file__).parent.absolute() / "wecker.pid") \ No newline at end of file diff --git a/tests/test_api.py b/tests/test_api.py index 7448b4d..9050382 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -1,3 +1,4 @@ +from unittest.mock import patch from fastapi.testclient import TestClient import tempfile import os @@ -14,6 +15,33 @@ from api.main import app # noqa: E402 client = TestClient(app) +def test_is_ringing_returns_false_when_not_running(): + """Test that isRinging returns False when no wecker process is running.""" + headers = {"X-API-Key": "test-secret"} + query = """ + query { + isRinging + } + """ + res = client.post("/graphql", json={"query": query}, headers=headers) + assert res.status_code == 200 + assert res.json()["data"]["isRinging"] is False + + +@patch("api.schema.is_wecker_ringing", return_value=True) +def test_is_ringing_returns_true_when_running(mock_is_ringing): + """Test that isRinging returns True when a wecker process is running.""" + headers = {"X-API-Key": "test-secret"} + query = """ + query { + isRinging + } + """ + res = client.post("/graphql", json={"query": query}, headers=headers) + assert res.status_code == 200 + assert res.json()["data"]["isRinging"] is True + + def test_auth_missing(): query = """ query { @@ -173,3 +201,13 @@ def test_graphql_workflow(): # 7. List again is empty res = client.post("/graphql", json={"query": query_get}, headers=headers) assert res.json()["data"]["getAlarms"] == [] + + +def test_pid_file_same_shared_constant_in_api(): + """Ensure api.schema uses the same PID_FILE from common, not a redefinition.""" + from api import schema as api_schema + import common + + assert api_schema.PID_FILE is common.PID_FILE, ( + "api.schema.PID_FILE must reference common.PID_FILE, not redefine it" + ) diff --git a/tests/test_single_instance.py b/tests/test_single_instance.py index 4dca977..c87980b 100644 --- a/tests/test_single_instance.py +++ b/tests/test_single_instance.py @@ -1,7 +1,7 @@ -import os -import sys import pytest -from unittest.mock import MagicMock, patch +from unittest.mock import patch + +import common import wecker # We need to mock the PID file existence and os.kill to test the logic @@ -80,3 +80,17 @@ def test_remove_pid_file_wrong_pid(mock_pid_file): wecker.remove_pid_file() assert 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 + 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 a6b0439..38deeb4 100644 --- a/wecker.py +++ b/wecker.py @@ -6,6 +6,8 @@ import random import sys import os +from common import PID_FILE + # Configure logging logging.basicConfig( level=logging.INFO, @@ -13,8 +15,6 @@ logging.basicConfig( handlers=[logging.FileHandler("wecker.log"), logging.StreamHandler()], ) -PID_FILE = "wecker.pid" - def ensure_single_instance(): """Ensures that only one instance of the script is running."""