changing tests, so the execution time is only checked for unix and cf (windows and slurm/dask can spend a lot of time to create directory etc. and how to set a good check)

Author: djarecka

pydra/engine/tests/test_workflow.py

@@ -1,5 +1,5 @@
 import pytest
-import shutil, os
+import shutil, os, sys
 import time
 import attr
 from pathlib import Path
@@ -2070,9 +2070,11 @@ def test_wf_nostate_cachelocations(plugin, tmpdir):
     results2 = wf2.result()
     assert 8 == results2.output.out
 
-    # checking execution time
-    assert t1 > 3
-    assert t2 < 0.5
+    # checking execution time (for unix and cf)
+    # for win and dask/slurm the time for dir creation etc. might take much longer
+    if sys.platform.startswith("win") and plugin == "cf":
+        assert t1 > 2
+        assert t2 < 0.5
 
     # checking if the second wf didn't run again
     assert wf1.output_dir.exists()
@@ -2125,10 +2127,12 @@ def test_wf_nostate_cachelocations_a(plugin, tmpdir):
     results2 = wf2.result()
     assert 8 == results2.output.out
 
-    # checking execution time (second one should be quick)
-    assert t1 > 3
-    # testing relative values (windows or slurm takes much longer to create wf itself)
-    assert t2 / t1 < 0.9
+    # for win and dask/slurm the time for dir creation etc. might take much longer
+    if sys.platform.startswith("win") and plugin == "cf":
+        # checking execution time (second one should be quick)
+        assert t1 > 2
+        # testing relative values (windows or slurm takes much longer to create wf itself)
+        assert t2 < 0.5
 
     # checking if both wf.output_dir are created
     assert wf1.output_dir.exists()
@@ -2184,9 +2188,11 @@ def test_wf_nostate_cachelocations_b(plugin, tmpdir):
     results2 = wf2.result()
     assert 8 == results2.output.out == results2.output.out_pr
 
-    # checking execution time
-    assert t1 > 3
-    assert t2 / t1 < 0.9
+    # for win and dask/slurm the time for dir creation etc. might take much longer
+    if sys.platform.startswith("win") and plugin == "cf":
+        # checking execution time
+        assert t1 > 2
+        assert t2 < 0.5
 
     # checking if the second wf didn't run again
     assert wf1.output_dir.exists()
@@ -2240,10 +2246,12 @@ def test_wf_nostate_cachelocations_setoutputchange(plugin, tmpdir):
     results2 = wf2.result()
     assert 8 == results2.output.out2
 
-    # checking execution time (the second wf should be fast, nodes do not have to rerun)
-    assert t1 > 3
-    # testing relative values (windows or slurm takes much longer to create wf itself)
-    assert t2 / t1 < 0.9
+    # for win and dask/slurm the time for dir creation etc. might take much longer
+    if sys.platform.startswith("win") and plugin == "cf":
+        # checking execution time (the second wf should be fast, nodes do not have to rerun)
+        assert t1 > 2
+        # testing relative values (windows or slurm takes much longer to create wf itself)
+        assert t2 < 0.5
 
     # both wf output_dirs should be created
     assert wf1.output_dir.exists()
@@ -2294,10 +2302,12 @@ def test_wf_nostate_cachelocations_setoutputchange_a(plugin, tmpdir):
     results2 = wf2.result()
     assert 8 == results2.output.out2
 
-    # checking execution time (the second wf should be fast, nodes do not have to rerun)
-    assert t1 > 3
-    # testing relative values (windows or slurm takes much longer to create wf itself)
-    assert t2 / t1 < 0.9
+    # for win and dask/slurm the time for dir creation etc. might take much longer
+    if sys.platform.startswith("win") and plugin == "cf":
+        # checking execution time (the second wf should be fast, nodes do not have to rerun)
+        assert t1 > 2
+        # testing relative values (windows or slurm takes much longer to create wf itself)
+        assert t2 < 0.5
 
     # both wf output_dirs should be created
     assert wf1.output_dir.exists()
@@ -2350,9 +2360,11 @@ def test_wf_nostate_cachelocations_forcererun(plugin, tmpdir):
     results2 = wf2.result()
     assert 8 == results2.output.out
 
-    # checking execution time
-    assert t1 > 3
-    assert t2 > 3
+    # for win and dask/slurm the time for dir creation etc. might take much longer
+    if sys.platform.startswith("win") and plugin == "cf":
+        # checking execution time
+        assert t1 > 2
+        assert t2 > 2
 
     # checking if the second wf didn't run again
     assert wf1.output_dir.exists()
@@ -2413,8 +2425,11 @@ def test_wf_nostate_cachelocations_wftaskrerun_propagateTrue(plugin, tmpdir):
     # everything has to be recomputed
     assert len(list(Path(cache_dir1).glob("F*"))) == 2
     assert len(list(Path(cache_dir2).glob("F*"))) == 2
-    assert t1 > 3
-    assert t2 > 3
+
+    # for win and dask/slurm the time for dir creation etc. might take much longer
+    if sys.platform.startswith("win") and plugin == "cf":
+        assert t1 > 2
+        assert t2 > 2
 
 
 def test_wf_nostate_cachelocations_wftaskrerun_propagateFalse(plugin, tmpdir):
@@ -2470,6 +2485,12 @@ def test_wf_nostate_cachelocations_wftaskrerun_propagateFalse(plugin, tmpdir):
     assert wf1.output_dir.exists()
     assert wf2.output_dir.exists()
 
+    # for win and dask/slurm the time for dir creation etc. might take much longer
+    if sys.platform.startswith("win") and plugin == "cf":
+        # checking the time
+        assert t1 > 2
+        assert t2 > 2
+
     # tasks should not be recomputed
     assert len(list(Path(cache_dir1).glob("F*"))) == 2
     assert len(list(Path(cache_dir2).glob("F*"))) == 0
@@ -2524,14 +2545,17 @@ def test_wf_nostate_cachelocations_taskrerun_wfrerun_propagateFalse(plugin, tmpd
     results2 = wf2.result()
     assert 8 == results2.output.out
 
+    # for win and dask/slurm the time for dir creation etc. might take much longer
+    if sys.platform.startswith("win") and plugin == "cf":
+        # checking the execution time
+        assert t1 > 2
+        assert t2 > 2
     # checking if the second wf doesn't runs again
     assert wf1.output_dir.exists()
     assert wf2.output_dir.exists()
     # the second task should be recomputed
     assert len(list(Path(cache_dir1).glob("F*"))) == 2
     assert len(list(Path(cache_dir2).glob("F*"))) == 1
-    assert t1 > 3
-    assert t2 > 3
 
 
 def test_wf_nostate_nodecachelocations(plugin, tmpdir):
@@ -2578,6 +2602,11 @@ def test_wf_nostate_nodecachelocations(plugin, tmpdir):
     results2 = wf2.result()
     assert 12 == results2.output.out
 
+    # for win and dask/slurm the time for dir creation etc. might take much longer
+    if sys.platform.startswith("win") and plugin == "cf":
+        # checking the execution time
+        assert t1 > 2
+        assert t2 > 2
     # checking if the second wf runs again, but runs only one task
     assert wf1.output_dir.exists()
     assert wf2.output_dir.exists()
@@ -2627,6 +2656,11 @@ def test_wf_nostate_nodecachelocations_upd(plugin, tmpdir):
     results2 = wf2.result()
     assert 12 == results2.output.out
 
+    # for win and dask/slurm the time for dir creation etc. might take much longer
+    if sys.platform.startswith("win") and plugin == "cf":
+        # checking the execution time
+        assert t1 > 2
+        assert t2 > 2
     # checking if the second wf runs again, but runs only one task
     assert wf1.output_dir.exists()
     assert wf2.output_dir.exists()
@@ -2684,9 +2718,11 @@ def test_wf_state_cachelocations(plugin, tmpdir):
     assert results2[0].output.out == 8
     assert results2[1].output.out == 82
 
-    # checking execution time
-    assert t1 > 3
-    assert t2 < 0.5
+    # for win and dask/slurm the time for dir creation etc. might take much longer
+    if sys.platform.startswith("win") and plugin == "cf":
+        # checking the execution time
+        assert t1 > 2
+        assert t2 < 0.5
 
     # checking all directories
     assert wf1.output_dir
@@ -2749,9 +2785,11 @@ def test_wf_state_cachelocations_forcererun(plugin, tmpdir):
     assert results2[0].output.out == 8
     assert results2[1].output.out == 82
 
-    # checking execution time
-    assert t1 > 3
-    assert t2 > 3
+    # for win and dask/slurm the time for dir creation etc. might take much longer
+    if sys.platform.startswith("win") and plugin == "cf":
+        # checking the execution time
+        assert t1 > 2
+        assert t2 > 2
 
     # checking all directories
     assert wf1.output_dir
@@ -2816,9 +2854,11 @@ def test_wf_state_cachelocations_updateinp(plugin, tmpdir):
     assert results2[0].output.out == 8
     assert results2[1].output.out == 82
 
-    # checking execution time
-    assert t1 > 3
-    assert t2 < 0.5
+    # for win and dask/slurm the time for dir creation etc. might take much longer
+    if sys.platform.startswith("win") and plugin == "cf":
+        # checking the execution time
+        assert t1 > 2
+        assert t2 < 0.5
 
     # checking all directories
     assert wf1.output_dir
@@ -2930,9 +2970,11 @@ def test_wf_nostate_cachelocations_updated(plugin, tmpdir):
     results2 = wf2.result()
     assert 8 == results2.output.out
 
-    # checking execution time
-    assert t1 > 3
-    assert t2 > 3
+    # for win and dask/slurm the time for dir creation etc. might take much longer
+    if sys.platform.startswith("win") and plugin == "cf":
+        # checking the execution time
+        assert t1 > 2
+        assert t2 > 2
 
     # checking if both wf run
     assert wf1.output_dir.exists()
@@ -2990,6 +3032,12 @@ def test_wf_nostate_cachelocations_recompute(plugin, tmpdir):
     assert wf1.output_dir.exists()
     assert wf2.output_dir.exists()
 
+    # for win and dask/slurm the time for dir creation etc. might take much longer
+    if sys.platform.startswith("win") and plugin == "cf":
+        # checking the execution time
+        assert t1 > 2
+        assert t2 > 2
+
     # the second wf should have only one task run
     assert len(list(Path(cache_dir1).glob("F*"))) == 2
     assert len(list(Path(cache_dir2).glob("F*"))) == 1
@@ -3044,9 +3092,11 @@ def test_wf_ndstate_cachelocations(plugin, tmpdir):
     results2 = wf2.result()
     assert results2.output.out == [8, 82]
 
-    # checking execution time
-    assert t1 > 3
-    assert t2 < 0.5
+    # for win and dask/slurm the time for dir creation etc. might take much longer
+    if sys.platform.startswith("win") and plugin == "cf":
+        # checking the execution time
+        assert t1 > 2
+        assert t2 < 0.5
 
     # checking all directories
     assert wf1.output_dir.exists()
@@ -3106,9 +3156,11 @@ def test_wf_ndstate_cachelocations_forcererun(plugin, tmpdir):
     results2 = wf2.result()
     assert results2.output.out == [8, 82]
 
-    # checking execution time
-    assert t1 > 3
-    assert t2 > 3
+    # for win and dask/slurm the time for dir creation etc. might take much longer
+    if sys.platform.startswith("win") and plugin == "cf":
+        # checking the execution time
+        assert t1 > 2
+        assert t2 > 2
 
     # checking all directories
     assert wf1.output_dir.exists()
@@ -3166,9 +3218,11 @@ def test_wf_ndstate_cachelocations_updatespl(plugin, tmpdir):
     results2 = wf2.result()
     assert results2.output.out == [8, 82]
 
-    # checking execution time
-    assert t1 > 3
-    assert t2 < 0.5
+    # for win and dask/slurm the time for dir creation etc. might take much longer
+    if sys.platform.startswith("win") and plugin == "cf":
+        # checking the execution time
+        assert t1 > 2
+        assert t2 < 0.5
 
     # checking all directories
     assert wf1.output_dir.exists()
@@ -3227,9 +3281,11 @@ def test_wf_ndstate_cachelocations_recompute(plugin, tmpdir):
     results2 = wf2.result()
     assert results2.output.out == [8, 10, 62, 82]
 
-    # checking execution time
-    assert t1 > 3
-    assert t2 > 3
+    # for win and dask/slurm the time for dir creation etc. might take much longer
+    if sys.platform.startswith("win") and plugin == "cf":
+        # checking the execution time
+        assert t1 > 2
+        assert t2 > 2
 
     # checking all directories
     assert wf1.output_dir.exists()
@@ -3254,8 +3310,10 @@ def test_wf_nostate_runtwice_usecache(plugin, tmpdir):
     wf1.inputs.y = 3
     wf1.plugin = plugin
 
+    t0 = time.time()
     with Submitter(plugin=plugin) as sub:
         sub(wf1)
+    t1 = time.time() - t0
 
     results1 = wf1.result()
     assert 8 == results1.output.out
@@ -3266,14 +3324,22 @@ def test_wf_nostate_runtwice_usecache(plugin, tmpdir):
     cache_dir_content = os.listdir(wf1.cache_dir)
 
     # running workflow the second time
+    t0 = time.time()
     with Submitter(plugin=plugin) as sub:
         sub(wf1)
+    t2 = time.time() - t0
 
     results1 = wf1.result()
     assert 8 == results1.output.out
-    # checking if no new directory is not created
+    # checking if no new directory is created
     assert cache_dir_content == os.listdir(wf1.cache_dir)
 
+    # for win and dask/slurm the time for dir creation etc. might take much longer
+    if sys.platform.startswith("win") and plugin == "cf":
+        # checking the execution time
+        assert t1 > 2
+        assert t2 < 0.5
+
 
 def test_wf_state_runtwice_usecache(plugin, tmpdir):
     """
@@ -3291,8 +3357,10 @@ def test_wf_state_runtwice_usecache(plugin, tmpdir):
     wf1.inputs.y = [3, 30]
     wf1.plugin = plugin
 
+    t0 = time.time()
     with Submitter(plugin=plugin) as sub:
         sub(wf1)
+    t1 = time.time() - t0
 
     results1 = wf1.result()
     assert 8 == results1[0].output.out
@@ -3305,14 +3373,21 @@ def test_wf_state_runtwice_usecache(plugin, tmpdir):
     cache_dir_content = os.listdir(wf1.cache_dir)
 
     # running workflow the second time
+    t0 = time.time()
     with Submitter(plugin=plugin) as sub:
         sub(wf1)
+    t2 = time.time() - t0
 
     results1 = wf1.result()
     assert 8 == results1[0].output.out
     assert 602 == results1[1].output.out
-    # checking if no new directory is not created
+    # checking if no new directory is created
     assert cache_dir_content == os.listdir(wf1.cache_dir)
+    # for win and dask/slurm the time for dir creation etc. might take much longer
+    if sys.platform.startswith("win") and plugin == "cf":
+        # checking the execution time
+        assert t1 > 2
+        assert t2 < 0.5
 
 
 @pytest.fixture

pydra/engine/tests/utils.py

@@ -128,7 +128,7 @@ def ten(x):
 
 @mark.task
 def add2_wait(x):
-    time.sleep(3)
+    time.sleep(2)
     return x + 2