feat: add startWecker and stopWecker mutations to start/stop alarm via API

- Add startWecker mutation: starts wecker.py via subprocess if not already ringing
- Add stopWecker mutation: kills the running wecker process via SIGTERM if ringing
- Both mutations handle the 'already ringing' / 'not ringing' edge cases gracefully
- Update README with API documentation for the new mutations
- Add comprehensive tests (unit + GraphQL endpoint)
This commit is contained in:
Markus Graf
2026-05-19 15:22:48 +02:00
parent 558c4ff5b6
commit df7bec1054
3 changed files with 149 additions and 0 deletions
+15
View File
@@ -99,6 +99,21 @@ mutation {
} }
``` ```
**Start the alarm immediately:**
```graphql
mutation {
startWecker
}
```
Returns `true` if the alarm started, `false` if it was already ringing (ignored).
**Stop the alarm immediately:**
```graphql
mutation {
stopWecker
}
```
Returns `true` if the alarm was stopped, `false` if it wasn't ringing.
### Example response for `isRinging`: ### Example response for `isRinging`:
```json ```json
+33
View File
@@ -2,6 +2,8 @@ import strawberry
from typing import List, Optional from typing import List, Optional
import os import os
import sys import sys
import subprocess
import signal
from pathlib import Path from pathlib import Path
from api.crontab_manager import CrontabManager from api.crontab_manager import CrontabManager
from common import PID_FILE from common import PID_FILE
@@ -95,5 +97,36 @@ class Mutation:
return True return True
return False return False
@strawberry.field
def start_wecker(self) -> bool:
"""Start the wecker alarm if it is not already ringing.
Returns True if started, False if already ringing."""
if is_wecker_ringing():
return False
project_root = Path(__file__).parent.parent.absolute()
python_exec = sys.executable
cmd = f"cd {project_root} && {python_exec} wecker.py >> wecker.log 2>&1"
subprocess.Popen(cmd, shell=True)
return True
@strawberry.field
def stop_wecker(self) -> bool:
"""Stop the wecker alarm if it is currently ringing.
Returns True if stopped, False if not ringing."""
if not is_wecker_ringing():
return False
try:
with open(PID_FILE) as f:
pid = int(f.read().strip())
os.kill(pid, signal.SIGTERM)
if os.path.exists(PID_FILE):
os.remove(PID_FILE)
return True
except (ValueError, ProcessLookupError, PermissionError, OSError):
# PID file is stale or process already dead
if os.path.exists(PID_FILE):
os.remove(PID_FILE)
return False
schema = strawberry.Schema(query=Query, mutation=Mutation) schema = strawberry.Schema(query=Query, mutation=Mutation)
+101
View File
@@ -203,6 +203,107 @@ def test_graphql_workflow():
assert res.json()["data"]["getAlarms"] == [] assert res.json()["data"]["getAlarms"] == []
def test_start_wecker_starts_process_when_not_ringing():
"""startWecker returns True and spawns wecker.py when not already ringing."""
from api.schema import Mutation
mutation = Mutation()
with patch("api.schema.is_wecker_ringing", return_value=False), \
patch("subprocess.Popen") as mock_popen:
mock_proc = mock_popen.return_value
mock_proc.pid = 9999
result = mutation.start_wecker()
assert result is True
mock_popen.assert_called_once()
# Verify the command contains wecker.py
args, kwargs = mock_popen.call_args
assert "wecker.py" in args[0]
assert kwargs.get("shell") is True
def test_start_wecker_ignores_when_already_ringing():
"""startWecker returns False when wecker is already ringing."""
from api.schema import Mutation
mutation = Mutation()
with patch("api.schema.is_wecker_ringing", return_value=True), \
patch("api.schema.subprocess.Popen") as mock_popen:
result = mutation.start_wecker()
assert result is False
mock_popen.assert_not_called()
def test_stop_wecker_kills_process_when_ringing():
"""stopWecker returns True and kills the process wecker is ringing."""
from api.schema import Mutation
mutation = Mutation()
with patch("api.schema.is_wecker_ringing", return_value=True), \
patch("api.schema.os.kill") as mock_kill, \
patch("common.PID_FILE", "/tmp/fake_wecker.pid"), \
patch("builtins.open") as mock_open:
mock_f = mock_open.return_value.__enter__.return_value
mock_f.read.return_value = "1234"
mock_kill.return_value = None
result = mutation.stop_wecker()
assert result is True
mock_kill.assert_called_with(1234, 15)
def test_stop_wecker_does_nothing_when_not_ringing():
"""stopWecker returns False when wecker is not ringing."""
from api.schema import Mutation
mutation = Mutation()
with patch("api.schema.is_wecker_ringing", return_value=False), \
patch("api.schema.os.kill") as mock_kill:
result = mutation.stop_wecker()
assert result is False
mock_kill.assert_not_called()
def test_start_wecker_graphql_endpoint():
"""GraphQL mutation startWecker works via the API."""
headers = {"X-API-Key": "test-secret"}
mutation_str = """
mutation {
startWecker
}
"""
with patch("api.schema.is_wecker_ringing", return_value=False), \
patch("subprocess.Popen") as mock_popen:
mock_proc = mock_popen.return_value
mock_proc.pid = 9999
res = client.post("/graphql", json={"query": mutation_str}, headers=headers)
assert res.status_code == 200
assert res.json()["data"]["startWecker"] is True
def test_stop_wecker_graphql_endpoint():
"""GraphQL mutation stopWecker works via the API."""
headers = {"X-API-Key": "test-secret"}
mutation_str = """
mutation {
stopWecker
}
"""
with patch("api.schema.is_wecker_ringing", return_value=True), \
patch("api.schema.os.kill") as mock_kill, \
patch("builtins.open") as mock_open:
mock_f = mock_open.return_value.__enter__.return_value
mock_f.read.return_value = "1234"
mock_kill.return_value = None
res = client.post("/graphql", json={"query": mutation_str}, headers=headers)
assert res.status_code == 200
assert res.json()["data"]["stopWecker"] is True
def test_pid_file_same_shared_constant_in_api(): def test_pid_file_same_shared_constant_in_api():
"""Ensure api.schema uses the same PID_FILE from common, not a redefinition.""" """Ensure api.schema uses the same PID_FILE from common, not a redefinition."""
from api import schema as api_schema from api import schema as api_schema