feat: pass alarm style through crontab manager and GraphQL API
- crontab_manager.get_alarms now parses --style from the cron command (defaults to 'blink' for legacy entries without --style) - GraphQL Alarm type gains a 'style' field - setAlarm accepts a 'style' arg (default 'simple') and validates it against the styles registry; unknown styles raise a GraphQL error - _default_command embeds --style <name> into the cron command - startRinging accepts 'style' (default 'simple') and passes --style to the spawned wecker.py The crontab remains the single source of truth; the style simply rides in the cron command alongside --music-file.
This commit is contained in:
@@ -1,6 +1,9 @@
|
||||
import shlex
|
||||
import uuid
|
||||
from crontab import CronTab, CronSlices
|
||||
|
||||
from styles import LEGACY_STYLE
|
||||
|
||||
|
||||
class CrontabManager:
|
||||
COMMENT_PREFIX = "wecker-alarm:"
|
||||
@@ -28,6 +31,7 @@ class CrontabManager:
|
||||
"cron_expression": str(job.slices),
|
||||
"command": job.command,
|
||||
"is_enabled": job.is_enabled(),
|
||||
"style": _parse_style(job.command),
|
||||
}
|
||||
)
|
||||
return alarms
|
||||
@@ -64,3 +68,21 @@ class CrontabManager:
|
||||
comment = f"{self.COMMENT_PREFIX}{alarm_id}"
|
||||
cron.remove_all(comment=comment)
|
||||
cron.write()
|
||||
|
||||
|
||||
def _parse_style(command: str) -> str:
|
||||
"""Extract the --style value from a cron command.
|
||||
|
||||
Returns LEGACY_STYLE ('blink') for entries that predate --style, so an
|
||||
upgrade never silently changes an existing alarm's behaviour.
|
||||
"""
|
||||
try:
|
||||
tokens = shlex.split(command)
|
||||
except ValueError:
|
||||
return LEGACY_STYLE
|
||||
for i, tok in enumerate(tokens):
|
||||
if tok == "--style" and i + 1 < len(tokens):
|
||||
return tokens[i + 1]
|
||||
if tok.startswith("--style="):
|
||||
return tok.split("=", 1)[1]
|
||||
return LEGACY_STYLE
|
||||
|
||||
+15
-7
@@ -1,6 +1,7 @@
|
||||
import strawberry
|
||||
from typing import List, Optional
|
||||
import os
|
||||
import shlex
|
||||
import sys
|
||||
import time
|
||||
import subprocess
|
||||
@@ -8,6 +9,7 @@ import signal
|
||||
from pathlib import Path
|
||||
from api.crontab_manager import CrontabManager
|
||||
from common import PID_FILE
|
||||
from styles import get_style
|
||||
|
||||
|
||||
def get_manager():
|
||||
@@ -42,7 +44,7 @@ def _project_root() -> Path:
|
||||
return start.parent.parent
|
||||
|
||||
|
||||
def _default_command() -> str:
|
||||
def _default_command(style: str) -> str:
|
||||
"""Return the default shell command to run wecker.py from crontab."""
|
||||
project_root = _project_root()
|
||||
python_exec = sys.executable
|
||||
@@ -50,18 +52,18 @@ def _default_command() -> str:
|
||||
return (
|
||||
f"cd {project_root} && "
|
||||
f"XDG_RUNTIME_DIR={runtime_dir} SDL_AUDIODRIVER=pulse "
|
||||
f"{python_exec} wecker.py >> wecker.log 2>&1"
|
||||
f"{python_exec} wecker.py --style {shlex.quote(style)} >> wecker.log 2>&1"
|
||||
)
|
||||
|
||||
|
||||
def _start_wecker_process() -> subprocess.Popen:
|
||||
def _start_wecker_process(style: str) -> subprocess.Popen:
|
||||
"""Start wecker.py without invoking a shell."""
|
||||
project_root = _project_root()
|
||||
env = os.environ.copy()
|
||||
env.setdefault("SDL_AUDIODRIVER", "pulse")
|
||||
env.setdefault("XDG_RUNTIME_DIR", f"/run/user/{os.getuid()}")
|
||||
return subprocess.Popen(
|
||||
[sys.executable, str(project_root / "wecker.py")],
|
||||
[sys.executable, str(project_root / "wecker.py"), "--style", style],
|
||||
cwd=project_root,
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
@@ -75,6 +77,7 @@ class Alarm:
|
||||
id: str
|
||||
cron_expression: str
|
||||
is_enabled: bool
|
||||
style: str
|
||||
|
||||
|
||||
def _alarm_from_dict(data: dict) -> Alarm:
|
||||
@@ -83,6 +86,7 @@ def _alarm_from_dict(data: dict) -> Alarm:
|
||||
id=data["id"],
|
||||
cron_expression=data["cron_expression"],
|
||||
is_enabled=data["is_enabled"],
|
||||
style=data["style"],
|
||||
)
|
||||
|
||||
|
||||
@@ -118,8 +122,10 @@ class Mutation:
|
||||
cron_expression: str,
|
||||
is_enabled: bool = True,
|
||||
id: Optional[str] = None,
|
||||
style: str = "simple",
|
||||
) -> Alarm:
|
||||
command = _default_command()
|
||||
get_style(style) # validate; raises ValueError on unknown style
|
||||
command = _default_command(style)
|
||||
|
||||
manager = get_manager()
|
||||
new_id = manager.set_alarm(
|
||||
@@ -132,6 +138,7 @@ class Mutation:
|
||||
id=new_id,
|
||||
cron_expression=cron_expression,
|
||||
is_enabled=is_enabled,
|
||||
style=style,
|
||||
)
|
||||
|
||||
@strawberry.field
|
||||
@@ -146,12 +153,13 @@ class Mutation:
|
||||
return False
|
||||
|
||||
@strawberry.field
|
||||
def start_ringing(self) -> bool:
|
||||
def start_ringing(self, style: str = "simple") -> 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
|
||||
_start_wecker_process()
|
||||
get_style(style) # validate; raises ValueError on unknown style
|
||||
_start_wecker_process(style)
|
||||
return True
|
||||
|
||||
@strawberry.field
|
||||
|
||||
@@ -2,6 +2,7 @@ from unittest.mock import patch, MagicMock
|
||||
from fastapi.testclient import TestClient
|
||||
import tempfile
|
||||
import os
|
||||
import pytest
|
||||
import signal
|
||||
|
||||
# We need to set the environment variable before importing the app
|
||||
@@ -403,3 +404,107 @@ def test_pid_file_same_shared_constant_in_api():
|
||||
assert api_schema.PID_FILE is common.PID_FILE, (
|
||||
"api.schema.PID_FILE must reference common.PID_FILE, not redefine it"
|
||||
)
|
||||
|
||||
|
||||
def _cleanup_alarm(alarm_id):
|
||||
headers = {"X-API-Key": "test-secret"}
|
||||
client.post(
|
||||
"/graphql",
|
||||
json={"query": f'mutation {{ deleteAlarm(id: "{alarm_id}") }}'},
|
||||
headers=headers,
|
||||
)
|
||||
|
||||
|
||||
def test_set_alarm_accepts_and_returns_style():
|
||||
headers = {"X-API-Key": "test-secret"}
|
||||
mutation = '''
|
||||
mutation {
|
||||
setAlarm(cronExpression: "0 9 * * *", style: "blink") {
|
||||
id
|
||||
style
|
||||
}
|
||||
}
|
||||
'''
|
||||
res = client.post("/graphql", json={"query": mutation}, headers=headers)
|
||||
assert res.status_code == 200
|
||||
alarm = res.json()["data"]["setAlarm"]
|
||||
assert alarm["style"] == "blink"
|
||||
_cleanup_alarm(alarm["id"])
|
||||
|
||||
|
||||
def test_set_alarm_default_style_is_simple():
|
||||
headers = {"X-API-Key": "test-secret"}
|
||||
mutation = '''
|
||||
mutation {
|
||||
setAlarm(cronExpression: "0 9 * * *") { id style }
|
||||
}
|
||||
'''
|
||||
res = client.post("/graphql", json={"query": mutation}, headers=headers)
|
||||
assert res.status_code == 200
|
||||
alarm = res.json()["data"]["setAlarm"]
|
||||
assert alarm["style"] == "simple"
|
||||
_cleanup_alarm(alarm["id"])
|
||||
|
||||
|
||||
def test_set_alarm_rejects_unknown_style():
|
||||
headers = {"X-API-Key": "test-secret"}
|
||||
mutation = '''
|
||||
mutation {
|
||||
setAlarm(cronExpression: "0 9 * * *", style: "nope") { id }
|
||||
}
|
||||
'''
|
||||
res = client.post("/graphql", json={"query": mutation}, headers=headers)
|
||||
assert res.status_code == 200
|
||||
assert "errors" in res.json()
|
||||
assert "Unknown alarm style" in str(res.json()["errors"])
|
||||
|
||||
|
||||
def test_get_alarms_includes_style():
|
||||
headers = {"X-API-Key": "test-secret"}
|
||||
set_mut = '''
|
||||
mutation { setAlarm(cronExpression: "0 9 * * *", style: "blink") { id } }
|
||||
'''
|
||||
res = client.post("/graphql", json={"query": set_mut}, headers=headers)
|
||||
alarm_id = res.json()["data"]["setAlarm"]["id"]
|
||||
|
||||
res = client.post(
|
||||
"/graphql", json={"query": "{ getAlarms { id style } }"}, headers=headers
|
||||
)
|
||||
alarms = res.json()["data"]["getAlarms"]
|
||||
assert any(a["id"] == alarm_id and a["style"] == "blink" for a in alarms)
|
||||
_cleanup_alarm(alarm_id)
|
||||
|
||||
|
||||
def test_start_ringing_passes_style():
|
||||
from api.schema import Mutation
|
||||
|
||||
mutation = Mutation()
|
||||
with patch("api.schema.is_wecker_ringing", return_value=False), \
|
||||
patch("subprocess.Popen") as mock_popen:
|
||||
mock_popen.return_value.pid = 9999
|
||||
assert mutation.start_ringing(style="blink") is True
|
||||
argv = mock_popen.call_args[0][0]
|
||||
assert "--style" in argv and "blink" in argv
|
||||
|
||||
|
||||
def test_start_ringing_default_style_is_simple():
|
||||
from api.schema import Mutation
|
||||
|
||||
mutation = Mutation()
|
||||
with patch("api.schema.is_wecker_ringing", return_value=False), \
|
||||
patch("subprocess.Popen") as mock_popen:
|
||||
mock_popen.return_value.pid = 9999
|
||||
mutation.start_ringing()
|
||||
argv = mock_popen.call_args[0][0]
|
||||
assert "simple" in argv
|
||||
|
||||
|
||||
def test_start_ringing_rejects_unknown_style():
|
||||
from api.schema import Mutation
|
||||
|
||||
mutation = Mutation()
|
||||
with patch("api.schema.is_wecker_ringing", return_value=False), \
|
||||
patch("subprocess.Popen") as mock_popen:
|
||||
with pytest.raises(ValueError, match="Unknown alarm style"):
|
||||
mutation.start_ringing(style="nope")
|
||||
mock_popen.assert_not_called()
|
||||
|
||||
@@ -72,3 +72,38 @@ def test_set_alarm_rejects_invalid_cron_expression(crontab_file):
|
||||
manager = CrontabManager(tabfile=crontab_file)
|
||||
with pytest.raises(ValueError, match="Invalid cron expression"):
|
||||
manager.set_alarm("test-id", "not-a-cron-expression", "cmd", True)
|
||||
|
||||
|
||||
def test_get_alarms_parses_style(crontab_file):
|
||||
manager = CrontabManager(tabfile=crontab_file)
|
||||
manager.set_alarm(
|
||||
alarm_id="style-id",
|
||||
cron_expression="30 7 * * *",
|
||||
command="python wecker.py --style simple",
|
||||
is_enabled=True,
|
||||
)
|
||||
alarms = manager.get_alarms()
|
||||
assert len(alarms) == 1
|
||||
assert alarms[0]["style"] == "simple"
|
||||
|
||||
|
||||
def test_get_alarms_parses_eq_style(crontab_file):
|
||||
manager = CrontabManager(tabfile=crontab_file)
|
||||
manager.set_alarm(
|
||||
alarm_id="style-eq-id",
|
||||
cron_expression="30 7 * * *",
|
||||
command="python wecker.py --style=blink",
|
||||
is_enabled=True,
|
||||
)
|
||||
assert manager.get_alarms()[0]["style"] == "blink"
|
||||
|
||||
|
||||
def test_get_alarms_legacy_command_defaults_to_blink(crontab_file):
|
||||
manager = CrontabManager(tabfile=crontab_file)
|
||||
manager.set_alarm(
|
||||
alarm_id="legacy-id",
|
||||
cron_expression="30 7 * * *",
|
||||
command="python wecker.py",
|
||||
is_enabled=True,
|
||||
)
|
||||
assert manager.get_alarms()[0]["style"] == "blink"
|
||||
|
||||
Reference in New Issue
Block a user