Test for --max_latency (#67)

Author: sylvinus

tests/tasks/general.py

@@ -1,9 +1,9 @@
-from time import sleep
 from mrq.task import Task
 from mrq.context import (log, retry_current_job, connections, get_current_config, get_current_job,
                          set_job_progress, subpool_map, queue_job, abort_current_job)
 import urllib2
 import json
+import time
 
 
 class Add(Task):
@@ -15,7 +15,7 @@ def run(self, params):
 
         if params.get("sleep", 0):
             log.info("sleeping", params.get("sleep", 0))
-            sleep(params.get("sleep", 0))
+            time.sleep(params.get("sleep", 0))
 
         return res
 
@@ -24,6 +24,13 @@ class TimeoutFromConfig(Add):
     pass
 
 
+class GetTime(Task):
+
+    def run(self, params):
+
+        return time.time()
+
+
 class Fetch(Task):
 
     def run(self, params):
@@ -53,7 +60,7 @@ def run(self, params):
             LEAKS.append(["1" for _ in range(params.get("size", 0))])
 
         if params.get("sleep", 0) > 0:
-            sleep(params.get("sleep", 0))
+            time.sleep(params.get("sleep", 0))
 
         return params.get("return")
 
@@ -80,7 +87,7 @@ class RaiseException(Task):
 
     def run(self, params):
 
-        sleep(params.get("sleep", 0))
+        time.sleep(params.get("sleep", 0))
 
         raise Exception(params.get("message", ""))
 
@@ -96,7 +103,7 @@ class ReturnParams(Task):
 
     def run(self, params):
 
-        sleep(params.get("sleep", 0))
+        time.sleep(params.get("sleep", 0))
 
         return params
 
@@ -107,7 +114,7 @@ def run(self, params):
 
         for i in range(1, 100):
             set_job_progress(0.01 * i, save=params["save"])
-            sleep(0.1)
+            time.sleep(0.1)
 
 
 class MongoInsert(Task):
@@ -118,7 +125,7 @@ def run(self, params):
             {"params": params}, manipulate=False)
 
         if params.get("sleep", 0) > 0:
-            sleep(params.get("sleep", 0))
+            time.sleep(params.get("sleep", 0))
 
         return params
 
@@ -141,7 +148,7 @@ def inner(self, x):
         if x == "exception":
             raise Exception(x)
 
-        sleep(x)
+        time.sleep(x)
 
         return x
 

tests/test_performance.py

@@ -2,6 +2,38 @@
 from mrq.queue import Queue
 import pytest
 
+@pytest.mark.parametrize(["p_max_latency", "p_min_observed_latency", "p_max_observed_latency"], [
+    [1, 0.03, 1],
+    [0.01, -1, 0.02]
+])
+def test_job_max_latency(worker, p_max_latency, p_min_observed_latency, p_max_observed_latency):
+
+    worker.start(flags=" --greenlets=1 --max_latency=%s" % (p_max_latency), trace=False)
+
+    def get_latency():
+        t = time.time()
+        return worker.send_task("tests.tasks.general.GetTime", {}) - t
+
+    # Warm up the worker
+    get_latency()
+
+    # This is the latency induced by our test system & general task work
+    # We're on the same machine so even in different processes time.time() should be pretty reliable
+    base_latency = get_latency()
+    print "Base latency: %ss" % base_latency
+
+    min_latency = min([get_latency() for _ in range(0, 5)])
+    print "FYI, min latency = %ss" % min_latency
+
+    # Sleep a while with an idle worker to make the poll interval go up
+    time.sleep(30)
+
+    latency = get_latency() - base_latency
+
+    print "Observed latency: %ss" % latency
+
+    assert p_min_observed_latency <= latency < p_max_observed_latency
+
 
 @pytest.mark.parametrize(["p_latency", "p_min", "p_max"], [
     [0, 0, 3],