feat: implement crontab manager for managing alarms

api/crontab_manager.py

@@ -0,0 +1,54 @@
+import uuid
+from crontab import CronTab
+
+class CrontabManager:
+    COMMENT_PREFIX = "wecker-alarm:"
+
+    def __init__(self, tabfile: str | None = None, user: bool | str = True):
+        # user=True means current user, user="username" means specific user
+        self.tabfile = tabfile
+        self.user = user
+
+    def _get_cron(self):
+        if self.tabfile:
+            return CronTab(tabfile=self.tabfile)
+        return CronTab(user=self.user)
+
+    def get_alarms(self):
+        cron = self._get_cron()
+        alarms = []
+        for job in cron:
+            if job.comment.startswith(self.COMMENT_PREFIX):
+                alarm_id = job.comment.split(self.COMMENT_PREFIX)[1].strip()
+                # job.slices is a valid cron slice object, str(job.slices) gives the expression
+                alarms.append({
+                    "id": alarm_id,
+                    "cron_expression": str(job.slices),
+                    "command": job.command,
+                    "is_enabled": job.is_enabled()
+                })
+        return alarms
+
+    def set_alarm(self, alarm_id: str | None, cron_expression: str, command: str, is_enabled: bool = True):
+        if not alarm_id:
+            alarm_id = str(uuid.uuid4())
+            
+        cron = self._get_cron()
+        comment = f"{self.COMMENT_PREFIX}{alarm_id}"
+        
+        # Remove existing if any
+        cron.remove_all(comment=comment)
+        
+        # Create new
+        job = cron.new(command=command, comment=comment)
+        job.setall(cron_expression)
+        job.enable(is_enabled)
+        
+        cron.write()
+        return alarm_id
+
+    def delete_alarm(self, alarm_id: str):
+        cron = self._get_cron()
+        comment = f"{self.COMMENT_PREFIX}{alarm_id}"
+        cron.remove_all(comment=comment)
+        cron.write()

tests/test_crontab.py

@@ -0,0 +1,63 @@
+import pytest
+import tempfile
+import uuid
+from api.crontab_manager import CrontabManager
+
+@pytest.fixture
+def crontab_file():
+    with tempfile.NamedTemporaryFile(mode='w+', delete=False) as f:
+        pass
+    yield f.name
+
+def test_crontab_manager_empty(crontab_file):
+    manager = CrontabManager(tabfile=crontab_file)
+    assert manager.get_alarms() == []
+
+def test_add_and_list_alarm(crontab_file):
+    manager = CrontabManager(tabfile=crontab_file)
+    alarm_id = str(uuid.uuid4())
+    manager.set_alarm(
+        alarm_id=alarm_id,
+        cron_expression="30 7 * * *",
+        command="python wecker.py",
+        is_enabled=True
+    )
+    
+    alarms = manager.get_alarms()
+    assert len(alarms) == 1
+    assert alarms[0]["id"] == alarm_id
+    assert alarms[0]["cron_expression"] == "30 7 * * *"
+    assert alarms[0]["command"] == "python wecker.py"
+    assert alarms[0]["is_enabled"] is True
+
+def test_update_alarm(crontab_file):
+    manager = CrontabManager(tabfile=crontab_file)
+    alarm_id = "test-id"
+    manager.set_alarm(
+        alarm_id=alarm_id,
+        cron_expression="30 7 * * *",
+        command="python wecker.py",
+        is_enabled=True
+    )
+    
+    manager.set_alarm(
+        alarm_id=alarm_id,
+        cron_expression="0 8 * * *",
+        command="python wecker.py --loud",
+        is_enabled=False
+    )
+    
+    alarms = manager.get_alarms()
+    assert len(alarms) == 1
+    assert alarms[0]["cron_expression"] == "0 8 * * *"
+    assert alarms[0]["command"] == "python wecker.py --loud"
+    assert alarms[0]["is_enabled"] is False
+
+def test_delete_alarm(crontab_file):
+    manager = CrontabManager(tabfile=crontab_file)
+    alarm_id = "test-id-2"
+    manager.set_alarm(alarm_id, "0 0 * * *", "cmd", True)
+    assert len(manager.get_alarms()) == 1
+    
+    manager.delete_alarm(alarm_id)
+    assert len(manager.get_alarms()) == 0