CronAlarm: add exclude options (no UI)

fix #10

larigira/tests/test_time_cron.py

@@ -7,11 +7,17 @@ from larigira.timegen_cron import CronAlarm
 
 @pytest.fixture
 def a_time():
+    # 6th of august 2019 at 10:42
     return datetime.datetime(2019, 8, 6, 10, 42, 0)
 
 
-def CA(fmt):
-    return CronAlarm(dict(cron_format=fmt))
+@pytest.fixture(params=("* * * * *", "* * * * * *"))
+def valid_cron(request):
+    return request.param
+
+
+def CA(fmt, exclude=""):
+    return CronAlarm(dict(cron_format=fmt, exclude=exclude))
 
 
 def test_valid_cron_format():
@@ -22,7 +28,51 @@ def test_valid_cron_format_six():
     CA("* * * * * *")
 
 
+def test_valid_cron_format_spaces_left(valid_cron):
+    """if a  format is valid, a format with left spaces is also valid"""
+    CA(" " + valid_cron)
+
+
+def test_valid_cron_format_spaces_right(valid_cron):
+    """if a  format is valid, a format with right spaces is also valid"""
+    CA(valid_cron + " ")
+
+
+def test_invalid_cron_format_four():
+    with pytest.raises(ValueError):
+        CA("* * * *")
+
+
 def test_never_equal(a_time):
     c = CA("* * * * *")
     nt = c.next_ring(a_time)
     assert nt.minute != a_time.minute
+
+
+def test_exclude_single(valid_cron):
+    CA(valid_cron, valid_cron)
+
+
+def test_exclude_multi_newline(valid_cron):
+    CA(valid_cron, valid_cron + "\n" + valid_cron)
+
+
+def test_exclude_multi_list(valid_cron):
+    CA(valid_cron, [valid_cron, valid_cron])
+
+
+def test_exclude_works(a_time):
+    c = CA("* * * * *")
+    nt = c.next_ring(a_time)
+    assert nt.day == 6
+    c = CA("* * * * *", "* * 6 * *")
+    nt = c.next_ring(a_time)
+    assert nt is not None
+    assert nt.day == 7
+
+
+def test_exclude_fails(a_time):
+    """exclude fails if every specification in cron_format is excluded"""
+    c = CA("* * * * *", "* * * * *")
+    assert c.has_ring(a_time) is False
+    assert c.next_ring(a_time) is None

larigira/timegen_cron.py

@@ -1,5 +1,5 @@
 import logging
-from datetime import datetime
+from datetime import datetime, timedelta
 
 from croniter import croniter
 
@@ -16,15 +16,43 @@ class CronAlarm(Alarm):
         super().__init__()
 
         self.cron_format = obj["cron_format"]
+        if "exclude" in obj:
+            if type(obj["exclude"]) is str:
+                self.exclude = [
+                    line.strip() for line in obj["exclude"].split("\n") if line.strip()
+                ]
+            else:
+                self.exclude = obj["exclude"]
+        else:
+            self.exclude = []
         if not croniter.is_valid(self.cron_format):
             raise ValueError("Invalid cron_format: %s" % self.cron_format)
+        for exclude in self.exclude:
+            if not croniter.is_valid(exclude):
+                raise ValueError("Invalid exclude: %s" % exclude)
+
+    def is_excluded(self, dt):
+        base = dt - timedelta(seconds=1)
+        for exclude in self.exclude:
+            nt = croniter(exclude, base).get_next(datetime)
+            if nt == dt:
+                return True
+        return False
 
     def next_ring(self, current_time=None):
         if current_time is None:
             current_time = datetime.now()
 
-        return croniter(self.cron_format, current_time).get_next(datetime)
+        # cron granularity is to the minute
+        # thus, doing 2000 attemps guarantees at least 32hours.
+        # if your event is no more frequent than 10minutes, this is 13days
+        for _ in range(2000):
+            nt = croniter(self.cron_format, current_time).get_next(datetime)
+            if not self.is_excluded(nt):
+                return nt
+            current_time = nt
+        return None
 
     def has_ring(self, current_time=None):
         # cron specification has no possibility of being over
-        return True
+        return self.next_ring(current_time) is not None