Add a test for running subpool_map outside of a Job. Tag as 0.1.4

Author: sylvinus

mrq/context.py

@@ -1,5 +1,6 @@
 from .logger import LoggerInterface
 import gevent
+import gevent.pool
 import urlparse
 import re
 import time

tests/tasks/general.py

@@ -127,7 +127,8 @@ class SubPool(Task):
 
     def inner(self, x):
 
-        assert get_current_job() == self.job
+        if self.job:
+            assert get_current_job() == self.job
 
         if x == "import-large-file":
             from tests.tasks.largefile import a

tests/test_subpool.py

@@ -6,29 +6,43 @@
 import pytest
 
 
-def test_subpool_simple(worker):
+@pytest.mark.parametrize(["use_worker"], [[False], [True]])
+def test_subpool_simple(worker, use_worker):
 
-    worker.start()
+    # Check that a subpool can be used both in an outside of a Job context
+    if use_worker:
+        worker.start()
+    else:
+        from tests.tasks.general import SubPool
+
+    def run(params):
+        if use_worker:
+            return worker.send_task("tests.tasks.general.SubPool", params)
+        else:
+            return SubPool().run(params)
 
     # Check that sequential sleeps work
     start_time = time.time()
-    result = worker.send_task("tests.tasks.general.SubPool", {
+    result = run({
         "pool_size": 1, "inner_params": [1, 1]
     })
     total_time = time.time() - start_time
 
     assert result == [1, 1]
     assert total_time > 2
 
-    # Parallel sleeps
-    start_time = time.time()
-    result = worker.send_task("tests.tasks.general.SubPool", {
-        "pool_size": 20, "inner_params": [1] * 20
-    })
-    total_time = time.time() - start_time
+    # py.test doesn't use gevent so we don't get the benefits of the hub
+    if use_worker:
+
+        # Parallel sleeps
+        start_time = time.time()
+        result = run({
+            "pool_size": 20, "inner_params": [1] * 20
+        })
+        total_time = time.time() - start_time
 
-    assert result == [1] * 20
-    assert total_time < 2
+        assert result == [1] * 20
+        assert total_time < 2
 
 
 def test_subpool_exception(worker):