Merge pull request #168 from pricingassistant/scheduler-patch

Scheduler patch

Author: FlorianPerucki

mrq/scheduler.py

@@ -91,7 +91,8 @@ def check(self):
 
             interval = datetime.timedelta(seconds=task["interval"])
 
-            last_time = now - interval
+            if task["datelastqueued"] >= now:
+                continue
 
             if task.get("monthday", current_monthday) != current_monthday:
                 continue
@@ -103,19 +104,22 @@ def check(self):
                 if task["datelastqueued"].date() == now.date() or now.time() < task["dailytime"].time():
                     continue
 
-            if task["datelastqueued"] <= last_time:
+            # if we only have "interval" key
+            if all(k not in task for k in ["monthday", "weekday", "dailytime"]):
+                if now - task["datelastqueued"] < interval:
+                    continue
 
-                queue_job(
-                    task["path"],
-                    task.get("params") or {},
-                    queue=task.get("queue")
-                )
+            queue_job(
+                task["path"],
+                task.get("params") or {},
+                queue=task.get("queue")
+            )
 
-                self.collection.update({"_id": task["_id"]}, {"$set": {
-                    "datelastqueued": now
-                }})
+            self.collection.update({"_id": task["_id"]}, {"$set": {
+                "datelastqueued": now
+            }})
 
-                log.debug("Scheduler: queued %s" % _hash_task(task))
+            log.debug("Scheduler: queued %s" % _hash_task(task))
 
         # Make sure we never again execute a scheduler with the same exact second.
         time.sleep(1)

tests/test_scheduler.py

@@ -44,7 +44,7 @@ def test_scheduler_simple(worker, p_flags):
 
     collection.remove({})
 
-    scheduled_jobs.update_many({}, {"$set": {"datelastqueued": datetime.datetime.utcnow()}})
+    scheduled_jobs.update_many({}, {"$set": {"datelastqueued": datetime.datetime.utcnow() + timedelta(seconds=10)}})
 
     # Start with new config
     worker.start(
@@ -114,6 +114,26 @@ def _start(deps=True):
     assert collection.find({"params.b": "test"}).count() == 1
 
 
+def test_scheduler_dailytime_with_datelastqueued(worker):
+    now = time.time()
+    worker.start(
+        flags="--scheduler --config tests/fixtures/config-scheduler3.py",
+        env={
+            # We need to pass this in the environment so that each worker has the
+            # exact same hash
+            "MRQ_TEST_SCHEDULER_TIME": str(now - 100 - 3600 * 24)
+        })
+    time.sleep(7)
+    print(list(worker.mongodb_jobs.tests_inserts.find()))
+    assert len(list(worker.mongodb_jobs.tests_inserts.find())) == 2
+    # pretend first run was yesterday at time(now) + 200s
+    datelastqueued = datetime.datetime.fromtimestamp(now - 3600*24 + 200)
+    worker.mongodb_jobs.mrq_scheduled_jobs.update({"params.b": "test"}, {"$set": {'datelastqueued': datelastqueued}})
+    time.sleep(3)
+    # task is ran
+    assert len(list(worker.mongodb_jobs.tests_inserts.find())) == 3
+
+
 def test_scheduler_weekday_dailytime(worker):
     # Task is scheduled in 5 seconds
     worker.start(