diff --git a/api/crontab_manager.py b/api/crontab_manager.py index 5113128..bd8ead7 100644 --- a/api/crontab_manager.py +++ b/api/crontab_manager.py @@ -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 diff --git a/api/schema.py b/api/schema.py index 42f328f..a724dbb 100644 --- a/api/schema.py +++ b/api/schema.py @@ -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 diff --git a/tests/test_api.py b/tests/test_api.py index 8499ebf..696d45c 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -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() diff --git a/tests/test_crontab.py b/tests/test_crontab.py index 8d97236..94c998d 100644 --- a/tests/test_crontab.py +++ b/tests/test_crontab.py @@ -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"