💚 Parallel tests

Makefile

@@ -59,7 +59,7 @@ lint: ## check style with flake8
 	flake8 tiatoolbox tests
 
 test: ## run tests quickly with the default Python
-	pytest
+	pytest -n auto
 
 coverage: ## check code coverage quickly with the default Python
 	pytest  --cov=tiatoolbox --cov-report=term --cov-report=html --cov-report=xml

requirements/requirements_dev.txt

@@ -10,6 +10,7 @@ pre-commit>=2.20.0
 pytest>=7.2.0
 pytest-cov>=4.0.0
 pytest-runner>=6.0
+pytest-xdist[psutil]
 ruff==0.0.284  # This will be updated by pre-commit bot to latest version
 toml>=0.10.2
 twine>=4.0.1

tests/models/test_hovernetplus.py

@@ -1,6 +1,5 @@
 """Unit test package for HoVerNet+."""
 
-from pathlib import Path
 from typing import Callable
 
 import torch
@@ -11,9 +10,8 @@
 from tiatoolbox.utils.transforms import imresize
 
 
-def test_functionality(remote_sample: Callable, tmp_path: Path) -> None:
+def test_functionality(remote_sample: Callable) -> None:
     """Functionality test."""
-    tmp_path = str(tmp_path)
     sample_patch = str(remote_sample("stainnorm-source"))
     patch_pre = imread(sample_patch)
     patch_pre = imresize(patch_pre, scale_factor=0.5)

tests/models/test_patch_predictor.py

@@ -22,6 +22,7 @@
 )
 from tiatoolbox.utils import download_data, imread, imwrite
 from tiatoolbox.utils import env_detection as toolbox_env
+from tiatoolbox.utils.misc import chdir
 from tiatoolbox.wsicore.wsireader import WSIReader
 
 ON_GPU = toolbox_env.has_gpu()
@@ -478,7 +479,7 @@ def test_io_patch_predictor_config() -> None:
 # -------------------------------------------------------------------------------------
 
 
-def test_predictor_crash() -> None:
+def test_predictor_crash(tmp_path) -> None:
     """Test for crash when making predictor."""
     # without providing any model
     with pytest.raises(ValueError, match=r"Must provide.*"):
@@ -496,20 +497,19 @@ def test_predictor_crash() -> None:
     predictor = PatchPredictor(pretrained_model="resnet18-kather100k", batch_size=32)
 
     with pytest.raises(ValueError, match=r".*not a valid mode.*"):
-        predictor.predict("aaa", mode="random")
+        predictor.predict("aaa", mode="random", save_dir=tmp_path)
     # remove previously generated data
-    if Path.exists(Path("output")):
-        _rm_dir("output")
+    _rm_dir(tmp_path / "output")
     with pytest.raises(TypeError, match=r".*must be a list of file paths.*"):
-        predictor.predict("aaa", mode="wsi")
+        predictor.predict("aaa", mode="wsi", save_dir=tmp_path)
     # remove previously generated data
-    _rm_dir("output")
+    _rm_dir(tmp_path / "output")
     with pytest.raises(ValueError, match=r".*masks.*!=.*imgs.*"):
-        predictor.predict([1, 2, 3], masks=[1, 2], mode="wsi")
+        predictor.predict([1, 2, 3], masks=[1, 2], mode="wsi", save_dir=tmp_path)
     with pytest.raises(ValueError, match=r".*labels.*!=.*imgs.*"):
-        predictor.predict([1, 2, 3], labels=[1, 2], mode="patch")
+        predictor.predict([1, 2, 3], labels=[1, 2], mode="patch", save_dir=tmp_path)
     # remove previously generated data
-    _rm_dir("output")
+    _rm_dir(tmp_path / "output")
 
 
 def test_io_config_delegation(remote_sample: Callable, tmp_path: Path) -> None:
@@ -629,34 +629,41 @@ def test_patch_predictor_api(sample_patch1, sample_patch2, tmp_path: Path) -> No
     output = predictor.predict(
         inputs,
         on_gpu=ON_GPU,
+        save_dir=save_dir_path,
     )
     assert sorted(output.keys()) == ["predictions"]
     assert len(output["predictions"]) == 2
+    _rm_dir(save_dir_path)
 
     output = predictor.predict(
         inputs,
         labels=[1, "a"],
         return_labels=True,
         on_gpu=ON_GPU,
+        save_dir=save_dir_path,
     )
     assert sorted(output.keys()) == sorted(["labels", "predictions"])
     assert len(output["predictions"]) == len(output["labels"])
     assert output["labels"] == [1, "a"]
+    _rm_dir(save_dir_path)
 
     output = predictor.predict(
         inputs,
         return_probabilities=True,
         on_gpu=ON_GPU,
+        save_dir=save_dir_path,
     )
     assert sorted(output.keys()) == sorted(["predictions", "probabilities"])
     assert len(output["predictions"]) == len(output["probabilities"])
+    _rm_dir(save_dir_path)
 
     output = predictor.predict(
         inputs,
         return_probabilities=True,
         labels=[1, "a"],
         return_labels=True,
         on_gpu=ON_GPU,
+        save_dir=save_dir_path,
     )
     assert sorted(output.keys()) == sorted(["labels", "predictions", "probabilities"])
     assert len(output["predictions"]) == len(output["labels"])
@@ -700,6 +707,7 @@ def test_patch_predictor_api(sample_patch1, sample_patch2, tmp_path: Path) -> No
         labels=[1, "a"],
         return_labels=True,
         on_gpu=ON_GPU,
+        save_dir=save_dir_path,
     )
     assert sorted(output.keys()) == sorted(["labels", "predictions", "probabilities"])
     assert len(output["predictions"]) == len(output["labels"])
@@ -718,6 +726,8 @@ def test_wsi_predictor_api(sample_wsi_dict, tmp_path: Path) -> None:
     patch_size = np.array([224, 224])
     predictor = PatchPredictor(pretrained_model="resnet18-kather100k", batch_size=32)
 
+    save_dir = f"{save_dir_path}/model_wsi_output"
+
     # wrapper to make this more clean
     kwargs = {
         "return_probabilities": True,
@@ -727,6 +737,7 @@ def test_wsi_predictor_api(sample_wsi_dict, tmp_path: Path) -> None:
         "stride_shape": patch_size,
         "resolution": 1.0,
         "units": "baseline",
+        "save_dir": save_dir,
     }
     # ! add this test back once the read at `baseline` is fixed
     # sanity check, both output should be the same with same resolution read args
@@ -737,6 +748,8 @@ def test_wsi_predictor_api(sample_wsi_dict, tmp_path: Path) -> None:
         **kwargs,
     )
 
+    _rm_dir(save_dir)
+
     tile_output = predictor.predict(
         [mini_wsi_jpg],
         masks=[mini_wsi_msk],
@@ -751,7 +764,6 @@ def test_wsi_predictor_api(sample_wsi_dict, tmp_path: Path) -> None:
     assert accuracy > 0.9, np.nonzero(~diff)
 
     # remove previously generated data
-    save_dir = f"{save_dir_path}/model_wsi_output"
     _rm_dir(save_dir)
 
     kwargs = {
@@ -800,26 +812,26 @@ def test_wsi_predictor_api(sample_wsi_dict, tmp_path: Path) -> None:
         )
     # remove previously generated data
     _rm_dir(_kwargs["save_dir"])
-    _rm_dir("output")
 
-    # test reading of multiple whole-slide images
-    _kwargs = copy.deepcopy(kwargs)
-    _kwargs["save_dir"] = None  # default coverage
-    _kwargs["return_probabilities"] = False
-    output = predictor.predict(
-        [mini_wsi_svs, mini_wsi_svs],
-        masks=[mini_wsi_msk, mini_wsi_msk],
-        mode="wsi",
-        **_kwargs,
-    )
-    assert Path.exists(Path("output"))
-    for output_info in output.values():
-        assert Path(output_info["raw"]).exists()
-        assert "merged" in output_info
-        assert Path(output_info["merged"]).exists()
+    with chdir(save_dir_path):
+        # test reading of multiple whole-slide images
+        _kwargs = copy.deepcopy(kwargs)
+        _kwargs["save_dir"] = None  # default coverage
+        _kwargs["return_probabilities"] = False
+        output = predictor.predict(
+            [mini_wsi_svs, mini_wsi_svs],
+            masks=[mini_wsi_msk, mini_wsi_msk],
+            mode="wsi",
+            **_kwargs,
+        )
+        assert Path.exists(Path("output"))
+        for output_info in output.values():
+            assert Path(output_info["raw"]).exists()
+            assert "merged" in output_info
+            assert Path(output_info["merged"]).exists()
 
-    # remove previously generated data
-    _rm_dir("output")
+        # remove previously generated data
+        _rm_dir("output")
 
 
 def test_wsi_predictor_merge_predictions(sample_wsi_dict) -> None:

tests/models/test_semantic_segmentation.py

@@ -29,6 +29,7 @@
 from tiatoolbox.models.models_abc import ModelABC
 from tiatoolbox.utils import env_detection as toolbox_env
 from tiatoolbox.utils import imread, imwrite
+from tiatoolbox.utils.misc import chdir
 from tiatoolbox.wsicore.wsireader import WSIReader
 
 ON_GPU = toolbox_env.has_gpu()
@@ -259,23 +260,25 @@ def test_functional_wsi_stream_dataset(remote_sample: Callable) -> None:
 # -------------------------------------------------------------------------------------
 
 
-def test_crash_segmentor(remote_sample: Callable) -> None:
+def test_crash_segmentor(remote_sample: Callable, tmp_path: Path) -> None:
     """Functional crash tests for segmentor."""
     # # convert to pathlib Path to prevent wsireader complaint
     mini_wsi_svs = Path(remote_sample("wsi2_4k_4k_svs"))
     mini_wsi_jpg = Path(remote_sample("wsi2_4k_4k_jpg"))
     mini_wsi_msk = Path(remote_sample("wsi2_4k_4k_msk"))
 
     model = _CNNTo1()
+
+    save_dir = Path(f"{tmp_path}/test_crash_segmentor")
     semantic_segmentor = SemanticSegmentor(batch_size=BATCH_SIZE, model=model)
+
     # fake injection to trigger Segmentor to create parallel
     # post processing workers because baseline Semantic Segmentor does not support
     # post processing out of the box. It only contains condition to create it
     # for any subclass
     semantic_segmentor.num_postproc_workers = 1
 
     # * test basic crash
-    _rm_dir("output")  # default output dir test
     with pytest.raises(TypeError, match=r".*`mask_reader`.*"):
         semantic_segmentor.filter_coordinates(mini_wsi_msk, np.array(["a", "b", "c"]))
     with pytest.raises(ValueError, match=r".*ndarray.*integer.*"):
@@ -292,7 +295,7 @@ def test_crash_segmentor(remote_sample: Callable) -> None:
             auto_get_mask=True,
         )
 
-    _rm_dir("output")  # default output dir test
+    _rm_dir(save_dir)
     with pytest.raises(ValueError, match=r".*provide.*"):
         SemanticSegmentor()
     with pytest.raises(ValueError, match=r".*valid mode.*"):
@@ -305,10 +308,16 @@ def test_crash_segmentor(remote_sample: Callable) -> None:
             mode="tile",
             on_gpu=ON_GPU,
             crash_on_exception=True,
+            save_dir=save_dir,
         )
     with pytest.raises(ValueError, match=r".*already exists.*"):
-        semantic_segmentor.predict([], mode="tile", patch_input_shape=(2048, 2048))
-    _rm_dir("output")  # default output dir test
+        semantic_segmentor.predict(
+            [],
+            mode="tile",
+            patch_input_shape=(2048, 2048),
+            save_dir=save_dir,
+        )
+    _rm_dir(save_dir)
 
     # * test not providing any io_config info when not using pretrained model
     with pytest.raises(ValueError, match=r".*provide either `ioconfig`.*"):
@@ -317,11 +326,11 @@ def test_crash_segmentor(remote_sample: Callable) -> None:
             mode="tile",
             on_gpu=ON_GPU,
             crash_on_exception=True,
+            save_dir=save_dir,
         )
-    _rm_dir("output")  # default output dir test
+    _rm_dir(save_dir)
 
     # * Test crash propagation when parallelize post processing
-    _rm_dir("output")
     semantic_segmentor.num_postproc_workers = 2
     semantic_segmentor.model.forward = _crash_func
     with pytest.raises(ValueError, match=r"Propagation Crash."):
@@ -331,17 +340,19 @@ def test_crash_segmentor(remote_sample: Callable) -> None:
             mode="wsi",
             on_gpu=ON_GPU,
             crash_on_exception=True,
+            save_dir=save_dir,
         )
-    _rm_dir("output")
+    _rm_dir(save_dir)
+
     # test ignore crash
     semantic_segmentor.predict(
         [mini_wsi_svs],
         patch_input_shape=(2048, 2048),
         mode="wsi",
         on_gpu=ON_GPU,
         crash_on_exception=False,
+        save_dir=save_dir,
     )
-    _rm_dir("output")
 
 
 def test_functional_segmentor_merging(tmp_path: Path) -> None:
@@ -467,7 +478,7 @@ def test_functional_segmentor(remote_sample: Callable, tmp_path: Path) -> None:
     imwrite(mini_wsi_msk, (thumb > 0).astype(np.uint8))
 
     # preemptive clean up
-    _rm_dir("output")  # default output dir test
+    _rm_dir(save_dir)  # default output dir test
     model = _CNNTo1()
     semantic_segmentor = SemanticSegmentor(batch_size=BATCH_SIZE, model=model)
     # fake injection to trigger Segmentor to create parallel
@@ -485,9 +496,10 @@ def test_functional_segmentor(remote_sample: Callable, tmp_path: Path) -> None:
         resolution=resolution,
         units="mpp",
         crash_on_exception=False,
+        save_dir=save_dir,
     )
 
-    _rm_dir("output")  # default output dir test
+    _rm_dir(save_dir)  # default output dir test
     semantic_segmentor.predict(
         [mini_wsi_jpg],
         mode="tile",
@@ -496,23 +508,25 @@ def test_functional_segmentor(remote_sample: Callable, tmp_path: Path) -> None:
         resolution=1 / resolution,
         units="baseline",
         crash_on_exception=True,
+        save_dir=save_dir,
     )
-    _rm_dir("output")  # default output dir test
+    _rm_dir(save_dir)  # default output dir test
 
-    # * check exception bypass in the log
-    # there should be no exception, but how to check the log?
-    semantic_segmentor.predict(
-        [mini_wsi_jpg],
-        mode="tile",
-        on_gpu=ON_GPU,
-        patch_input_shape=(512, 512),
-        patch_output_shape=(512, 512),
-        stride_shape=(512, 512),
-        resolution=1 / resolution,
-        units="baseline",
-        crash_on_exception=False,
-    )
-    _rm_dir("output")  # default output dir test
+    with chdir(tmp_path):
+        # * check exception bypass in the log
+        # there should be no exception, but how to check the log?
+        semantic_segmentor.predict(
+            [mini_wsi_jpg],
+            mode="tile",
+            on_gpu=ON_GPU,
+            patch_input_shape=(512, 512),
+            patch_output_shape=(512, 512),
+            stride_shape=(512, 512),
+            resolution=1 / resolution,
+            units="baseline",
+            crash_on_exception=False,
+        )
+        _rm_dir(tmp_path / "output")  # default output dir test
 
     # * test basic running and merging prediction
     # * should dumping all 1 in the output

tests/test_init.py

@@ -12,14 +12,14 @@
 from tiatoolbox import DuplicateFilter, logger
 
 
-def test_set_root_dir() -> None:
+def test_set_root_dir(tmp_path) -> None:
     """Test for setting new root dir."""
     # skipcq
     importlib.reload(tiatoolbox)
     from tiatoolbox import rcParam
 
     old_root_dir = rcParam["TIATOOLBOX_HOME"]
-    test_dir_path = Path.cwd() / "tmp_check"
+    test_dir_path = tmp_path / "tmp_check"
     # clean up previous test
     if Path.exists(test_dir_path):
         Path.rmdir(test_dir_path)