More test fixes

Author: sylvinus

mrq/monkey.py

@@ -99,9 +99,9 @@ def mrq_monkey_patched(self, *args, **kwargs):
                 ret = base_method(self, *args, **kwargs)
             finally:
                 stop_time = time.time()
-                
+
                 job = None
-                
+
                 if config["trace_io"]:
                     job = get_current_job()
                     if job:
@@ -388,6 +388,10 @@ class mrq_patched_pymongo_cursor(Cursor):
 
         # Some dark magic is needed here to cope with python's name mangling for private variables.
         def _Cursor__send_message(self, *args, **kwargs):
+
+            if len(args) and args[0].name == "find":
+                return Cursor._Cursor__send_message(self, *args, **kwargs)
+
             # print self.__dict__
             job = get_current_job()
 

tests/conftest.py

@@ -76,11 +76,11 @@ def start(self, cmdline=None, env=None, expected_children=0):
         if expected_children > 0:
             psutil_process = psutil.Process(self.process.pid)
 
-            # print "Expecting %s children, got %s" % (expected_children,
-            # psutil_process.get_children(recursive=False))
             while True:
-                self.process_children = psutil_process.children(
-                    recursive=True)
+                self.process_children = psutil_process.children(recursive=True)
+                # print("Expecting %s children of pid %s, got %s" % (
+                #     expected_children, self.process.pid, len(self.process_children))
+                # )
                 if len(self.process_children) >= expected_children:
                     break
                 time.sleep(0.1)
@@ -247,6 +247,11 @@ def get_report(self, with_memory=False):
         return data
 
     def get_wait_for_idle(self):
+
+        if "--processes" in self.cmdline:
+            print("Warning: get_wait_for_idle() doesn't support multiprocess workers yet")
+            return False
+
         try:
             wait_for_net_service("127.0.0.1", 20020, poll_interval=0.01)
             f = urllib.request.urlopen("http://127.0.0.1:20020/wait_for_idle")

tests/test_context.py

@@ -68,17 +68,18 @@ def test_context_metric_queue(worker):
 
     worker.start(flags=" --config tests/fixtures/config-metric.py")
 
+    # Will send 1 task inside!
     worker.send_task("tests.tasks.general.SendTask", {
                      "path": "tests.tasks.general.Add", "params": {"a": 41, "b": 1}})
 
     metrics = json.loads(
         worker.send_task("tests.tasks.general.GetMetrics", {}))
 
     # GetMetrics is also a task!
-    assert metrics.get("queues.default.dequeued") == 2
-    assert metrics.get("queues.all.dequeued") == 2
-    assert metrics.get("jobs.status.started") == 2
-    assert metrics.get("jobs.status.success") == 1  # At the time it is run, GetMetrics isn't success yet.
+    assert metrics.get("queues.default.dequeued") == 3
+    assert metrics.get("queues.all.dequeued") == 3
+    assert metrics.get("jobs.status.started") == 3
+    assert metrics.get("jobs.status.success") == 2  # At the time it is run, GetMetrics isn't success yet.
 
     TEST_LOCAL_METRICS.get("queues.default.enqueued") == 2
     TEST_LOCAL_METRICS.get("queues.all.enqueued") == 2

tests/test_io_hooks.py

@@ -74,11 +74,13 @@ def test_io_hooks_mongodb(worker):
 
     worker.start(flags=" --config tests/fixtures/config-io-hooks.py")
 
-    worker.send_task(
+    ret = worker.send_task(
         "tests.tasks.io.TestIo",
         {"test": "mongodb-full-getmore"}
     )
 
+    print(ret)
+
     events = json.loads(
         worker.send_task("tests.tasks.general.GetIoHookEvents", {}))
 
@@ -87,8 +89,6 @@ def test_io_hooks_mongodb(worker):
     for evt in job_events:
         print(evt)
 
-    assert len(job_events) == 4 * 2
-
     # First, insert
     assert job_events[0]["hook"] == "mongodb_pre"
     assert job_events[1]["hook"] == "mongodb_post"
@@ -119,13 +119,25 @@ def test_io_hooks_mongodb(worker):
     assert job_events[4]["method"] == "cursor"
     assert job_events[5]["method"] == "cursor"
 
-    # Result MongoDB update
-
+    # Then getmore query
     assert job_events[6]["hook"] == "mongodb_pre"
     assert job_events[7]["hook"] == "mongodb_post"
 
-    assert job_events[6]["method"] == "update"
-    assert job_events[7]["method"] == "update"
+    assert job_events[6]["collection"] == "mrq.tests_inserts"
+    assert job_events[7]["collection"] == "mrq.tests_inserts"
+
+    assert job_events[6]["method"] == "cursor"
+    assert job_events[7]["method"] == "cursor"
+
+    # Result MongoDB update
+
+    assert job_events[8]["hook"] == "mongodb_pre"
+    assert job_events[9]["hook"] == "mongodb_post"
+
+    assert job_events[8]["method"] == "update"
+    assert job_events[9]["method"] == "update"
+
+    assert job_events[8]["collection"] == "mrq.mrq_jobs"
+    assert job_events[9]["collection"] == "mrq.mrq_jobs"
 
-    assert job_events[6]["collection"] == "mrq.mrq_jobs"
-    assert job_events[7]["collection"] == "mrq.mrq_jobs"
+    assert len(job_events) == 5 * 2

tests/test_jobinspect.py

@@ -17,7 +17,7 @@ def test_current_job_inspect(worker):
     job_id = worker.send_task(
         "tests.tasks.general.MongoInsert", {"a": 41, "b": 1, "sleep": 3}, block=False)
 
-    time.sleep(1)
+    time.sleep(2)
 
     # Test the HTTP admin API
     admin_worker = json.loads(urllib.request.urlopen("http://localhost:20020").read().decode('utf-8'))

tests/test_parallel.py

@@ -12,6 +12,8 @@ def test_parallel_100sleeps(worker, p_flags):
 
     worker.start(flags=p_flags)
 
+    print("Worker started. Queueing sleeps")
+
     start_time = time.time()
 
     # This will sleep a total of 100 seconds
@@ -27,56 +29,48 @@ def test_parallel_100sleeps(worker, p_flags):
     assert result == list(range(100))
 
 
-@pytest.mark.parametrize(["p_greenlets"], [
-    [1],
-    [2]
+@pytest.mark.parametrize(["p_greenlets", "p_strategy"], [
+    [g, s]
+    for g in [1, 2]
+    for s in ["", "parallel", "burst"]
 ])
-def test_dequeue_strategy(worker, p_greenlets):
+def test_dequeue_strategy(worker, p_greenlets, p_strategy):
 
     worker.start_deps(flush=True)
 
     worker.send_task(
-        "tests.tasks.general.MongoInsert", {"a": 41, "sleep": 2}, queue="q1", block=False, start=False)
-    worker.send_task(
-        "tests.tasks.general.MongoInsert", {"a": 42, "sleep": 2}, queue="q2", block=False, start=False)
+        "tests.tasks.general.MongoInsert", {"a": 41, "sleep": 1}, queue="q1", block=False, start=False)
     worker.send_task(
-        "tests.tasks.general.MongoInsert", {"a": 41, "sleep": 2}, queue="q1", block=False, start=False)
+        "tests.tasks.general.MongoInsert", {"a": 42, "sleep": 1}, queue="q2", block=False, start=False)
     worker.send_task(
-        "tests.tasks.general.MongoInsert", {"a": 42, "sleep": 2}, queue="q2", block=False, start=False)
+        "tests.tasks.general.MongoInsert", {"a": 43, "sleep": 1}, queue="q1", block=False, start=False)
     worker.send_task(
-        "tests.tasks.general.MongoInsert", {"a": 43, "sleep": 2}, queue="q3", block=False, start=False)
+        "tests.tasks.general.MongoInsert", {"a": 44, "sleep": 1}, queue="q2", block=False, start=False)
     worker.send_task(
-        "tests.tasks.general.MongoInsert", {"a": 43, "sleep": 2}, queue="q3", block=False, start=False)
-
-    time.sleep(0.1)
-
-    worker.start(flags="--dequeue_strategy parallel --greenlets %s" % p_greenlets, queues="q1 q2", deps=False, start=False)
-
-    if p_greenlets == 1:
-        time.sleep(1 + 2)
-    else:
-        time.sleep(1)
+        "tests.tasks.general.MongoInsert", {"a": 45, "sleep": 1}, queue="q3", block=False, start=False)
 
-    # Should be dequeued in parallel
-    assert connections.mongodb_jobs.tests_inserts.count({"params.a": 41}) == 1
-    assert connections.mongodb_jobs.tests_inserts.count({"params.a": 42}) == 1
-    assert connections.mongodb_jobs.tests_inserts.count() == 2
+    time.sleep(0.5)
 
-    worker.stop(deps=False, sig=9)
-    time.sleep(1)
+    flags = "--greenlets %s" % p_greenlets
+    if p_strategy:
+        flags += " --dequeue_strategy %s" % p_strategy
 
-    worker.start(flags="--dequeue_strategy burst --greenlets 2", queues="q3", deps=False)
+    print("Worker has flags %s" % flags)
+    worker.start(flags=flags, queues="q1 q2", deps=False, block=False)
 
-    time.sleep(3)
+    gotit = worker.get_wait_for_idle()
 
-    assert connections.mongodb_jobs.tests_inserts.count({"params.a": 43}) == 2
-
-    # Worker should be stopped now so even if we queue nothing will happen.
-    worker.send_task(
-        "tests.tasks.general.MongoInsert", {"a": 43, "sleep": 2}, queue="q3", block=False, start=False)
-
-    time.sleep(2)
+    if p_strategy == "burst":
+        assert not gotit  # because worker should be stopped already
+    else:
+        assert gotit
 
-    assert connections.mongodb_jobs.tests_inserts.count({"params.a": 43}) == 2
+    inserts = list(connections.mongodb_jobs.tests_inserts.find(sort=[("_id", 1)]))
+    order = [row["params"]["a"] for row in inserts]
 
-    worker.stop()
+    if p_strategy == "parallel":
+        assert set(order[0:2]) == set([41, 42])
+        assert set(order[2:4]) == set([43, 44])
+    else:
+        assert set(order[0:2]) == set([41, 43])
+        assert set(order[2:4]) == set([42, 44])