🗑️ Implement clean up for test outputs (#960)

- Implement clean up of test outputs at the end of each run for efficiency.

Author: shaneahmed

tests/conftest.py

@@ -658,3 +658,39 @@ def timed(fn: Callable, *args: object) -> (Callable, float):
         end = time.time()
         compile_time = end - start
     return result, compile_time
+
+
+_tmp_paths: list[Path] = []
+
+
+@pytest.fixture
+def track_tmp_path(tmp_path: Path) -> Path:
+    """This fixture tracks `tmp_path` for clean up.
+
+    Fixture that wraps pytest's built-in `tmp_path` and tracks each temporary path
+    for later cleanup at the module level.
+
+    Returns:
+        Path: The temporary directory path for the current test function.
+
+    """
+    _tmp_paths.append(tmp_path)
+    return tmp_path
+
+
+@pytest.fixture(scope="module", autouse=True)
+def module_teardown() -> None:
+    """This module tears down temporary data directories.
+
+    Module-scoped fixture that automatically runs after all tests in a module.
+    It cleans up all temporary paths tracked during the module's execution.
+
+    Yields:
+        None: Allows pytest to run tests before executing the teardown logic.
+
+    """
+    yield
+    for path in _tmp_paths:
+        if path.exists():
+            shutil.rmtree(path)
+            print(f"Cleaned up: {path}")

tests/models/test_arch_micronet.py

@@ -54,7 +54,7 @@ def test_value_error() -> None:
     toolbox_env.running_on_ci() or not ON_GPU,
     reason="Local test on machine with GPU.",
 )
-def test_micronet_output(remote_sample: Callable, tmp_path: Path) -> None:
+def test_micronet_output(remote_sample: Callable, track_tmp_path: Path) -> None:
     """Test the output of MicroNet."""
     svs_1_small = Path(remote_sample("svs-1-small"))
     micronet_output = Path(remote_sample("micronet-output"))
@@ -74,7 +74,7 @@ def test_micronet_output(remote_sample: Callable, tmp_path: Path) -> None:
         imgs=[
             svs_1_small,
         ],
-        save_dir=tmp_path / "output",
+        save_dir=track_tmp_path / "output",
     )
 
     output = np.load(output[0][1] + ".raw.0.npy")

tests/models/test_dataset.py

@@ -86,9 +86,9 @@ def test_kather_nonexisting_dir() -> None:
         _ = KatherPatchDataset(save_dir_path="non-existing-path")
 
 
-def test_kather_dataset(tmp_path: Path) -> None:
+def test_kather_dataset(track_tmp_path: Path) -> None:
     """Test for kather patch dataset."""
-    save_dir_path = tmp_path
+    save_dir_path = track_tmp_path
 
     # save to temporary location
     # remove previously generated data

tests/models/test_feature_extractor.py

@@ -24,9 +24,9 @@
 # -------------------------------------------------------------------------------------
 
 
-def test_engine(remote_sample: Callable, tmp_path: Path) -> None:
+def test_engine(remote_sample: Callable, track_tmp_path: Path) -> None:
     """Test feature extraction with DeepFeatureExtractor engine."""
-    save_dir = tmp_path / "output"
+    save_dir = track_tmp_path / "output"
     # # convert to pathlib Path to prevent wsireader complaint
     mini_wsi_svs = Path(remote_sample("wsi4_1k_1k_svs"))
 
@@ -55,10 +55,10 @@ def test_engine(remote_sample: Callable, tmp_path: Path) -> None:
     "model", [CNNBackbone("resnet50"), TimmBackbone("efficientnet_b0", pretrained=True)]
 )
 def test_full_inference(
-    remote_sample: Callable, tmp_path: Path, model: Callable
+    remote_sample: Callable, track_tmp_path: Path, model: Callable
 ) -> None:
     """Test full inference with CNNBackbone and TimmBackbone models."""
-    save_dir = tmp_path / "output"
+    save_dir = track_tmp_path / "output"
     # pre-emptive clean up
     shutil.rmtree(save_dir, ignore_errors=True)  # default output dir test
 
@@ -121,9 +121,11 @@ def test_full_inference(
     toolbox_env.running_on_ci() or not ON_GPU,
     reason="Local test on machine with GPU.",
 )
-def test_multi_gpu_feature_extraction(remote_sample: Callable, tmp_path: Path) -> None:
+def test_multi_gpu_feature_extraction(
+    remote_sample: Callable, track_tmp_path: Path
+) -> None:
     """Local functionality test for feature extraction using multiple GPUs."""
-    save_dir = tmp_path / "output"
+    save_dir = track_tmp_path / "output"
     mini_wsi_svs = Path(remote_sample("wsi4_1k_1k_svs"))
     shutil.rmtree(save_dir, ignore_errors=True)
 

tests/models/test_multi_task_segmentor.py

@@ -48,10 +48,10 @@ def semantic_postproc_func(raw_output: np.ndarray) -> np.ndarray:
     toolbox_env.running_on_ci() or not ON_GPU,
     reason="Local test on machine with GPU.",
 )
-def test_functionality_local(remote_sample: Callable, tmp_path: Path) -> None:
+def test_functionality_local(remote_sample: Callable, track_tmp_path: Path) -> None:
     """Local functionality test for multi task segmentor."""
     gc.collect()
-    root_save_dir = Path(tmp_path)
+    root_save_dir = Path(track_tmp_path)
     mini_wsi_svs = Path(remote_sample("svs-1-small"))
     save_dir = root_save_dir / "multitask"
     shutil.rmtree(save_dir, ignore_errors=True)
@@ -100,9 +100,11 @@ def test_functionality_local(remote_sample: Callable, tmp_path: Path) -> None:
     assert score > 0.95, "Heavy loss of precision!"
 
 
-def test_functionality_hovernetplus(remote_sample: Callable, tmp_path: Path) -> None:
+def test_functionality_hovernetplus(
+    remote_sample: Callable, track_tmp_path: Path
+) -> None:
     """Functionality test for multitask segmentor."""
-    root_save_dir = Path(tmp_path)
+    root_save_dir = Path(track_tmp_path)
     mini_wsi_svs = Path(remote_sample("wsi4_512_512_svs"))
     required_dims = (258, 258)
     # above image is 512 x 512 at 0.252 mpp resolution. This is 258 x 258 at 0.500 mpp.
@@ -133,9 +135,9 @@ def test_functionality_hovernetplus(remote_sample: Callable, tmp_path: Path) ->
     )
 
 
-def test_functionality_hovernet(remote_sample: Callable, tmp_path: Path) -> None:
+def test_functionality_hovernet(remote_sample: Callable, track_tmp_path: Path) -> None:
     """Functionality test for multitask segmentor."""
-    root_save_dir = Path(tmp_path)
+    root_save_dir = Path(track_tmp_path)
     mini_wsi_svs = Path(remote_sample("wsi4_512_512_svs"))
 
     save_dir = root_save_dir / "multi"
@@ -159,14 +161,14 @@ def test_functionality_hovernet(remote_sample: Callable, tmp_path: Path) -> None
     assert len(inst_dict) > 0, "Must have some nuclei."
 
 
-def test_masked_segmentor(remote_sample: Callable, tmp_path: Path) -> None:
+def test_masked_segmentor(remote_sample: Callable, track_tmp_path: Path) -> None:
     """Test segmentor when image is masked."""
-    root_save_dir = Path(tmp_path)
+    root_save_dir = Path(track_tmp_path)
     sample_wsi_svs = Path(remote_sample("svs-1-small"))
     sample_wsi_msk = remote_sample("small_svs_tissue_mask")
     sample_wsi_msk = np.load(sample_wsi_msk).astype(np.uint8)
-    imwrite(f"{tmp_path}/small_svs_tissue_mask.jpg", sample_wsi_msk)
-    sample_wsi_msk = tmp_path.joinpath("small_svs_tissue_mask.jpg")
+    imwrite(f"{track_tmp_path}/small_svs_tissue_mask.jpg", sample_wsi_msk)
+    sample_wsi_msk = track_tmp_path.joinpath("small_svs_tissue_mask.jpg")
 
     save_dir = root_save_dir / "instance"
 
@@ -208,10 +210,10 @@ def test_masked_segmentor(remote_sample: Callable, tmp_path: Path) -> None:
 
 def test_functionality_process_instance_predictions(
     remote_sample: Callable,
-    tmp_path: Path,
+    track_tmp_path: Path,
 ) -> None:
     """Test the functionality of instance predictions processing."""
-    root_save_dir = Path(tmp_path)
+    root_save_dir = Path(track_tmp_path)
     mini_wsi_svs = Path(remote_sample("wsi4_512_512_svs"))
 
     save_dir = root_save_dir / "semantic"
@@ -251,9 +253,9 @@ def test_functionality_process_instance_predictions(
     assert len(multi_segmentor._wsi_inst_info[0]) == 0
 
 
-def test_empty_image(tmp_path: Path) -> None:
+def test_empty_image(track_tmp_path: Path) -> None:
     """Test MultiTaskSegmentor for an empty image."""
-    root_save_dir = Path(tmp_path)
+    root_save_dir = Path(track_tmp_path)
     sample_patch = np.ones((256, 256, 3), dtype="uint8") * 255
     sample_patch_path = root_save_dir / "sample_tile.png"
     imwrite(sample_patch_path, sample_patch)
@@ -320,9 +322,9 @@ def test_empty_image(tmp_path: Path) -> None:
     )
 
 
-def test_functionality_semantic(remote_sample: Callable, tmp_path: Path) -> None:
+def test_functionality_semantic(remote_sample: Callable, track_tmp_path: Path) -> None:
     """Functionality test for multitask segmentor."""
-    root_save_dir = Path(tmp_path)
+    root_save_dir = Path(track_tmp_path)
 
     save_dir = root_save_dir / "multi"
     shutil.rmtree(save_dir, ignore_errors=True)
@@ -373,14 +375,14 @@ def test_functionality_semantic(remote_sample: Callable, tmp_path: Path) -> None
     assert layer_map is not None, "Must have some segmentations."
 
 
-def test_crash_segmentor(remote_sample: Callable, tmp_path: Path) -> None:
+def test_crash_segmentor(remote_sample: Callable, track_tmp_path: Path) -> None:
     """Test engine crash when given malformed input."""
-    root_save_dir = Path(tmp_path)
+    root_save_dir = Path(track_tmp_path)
     sample_wsi_svs = Path(remote_sample("svs-1-small"))
     sample_wsi_msk = remote_sample("small_svs_tissue_mask")
     sample_wsi_msk = np.load(sample_wsi_msk).astype(np.uint8)
-    imwrite(f"{tmp_path}/small_svs_tissue_mask.jpg", sample_wsi_msk)
-    sample_wsi_msk = tmp_path.joinpath("small_svs_tissue_mask.jpg")
+    imwrite(f"{track_tmp_path}/small_svs_tissue_mask.jpg", sample_wsi_msk)
+    sample_wsi_msk = track_tmp_path.joinpath("small_svs_tissue_mask.jpg")
 
     save_dir = f"{root_save_dir}/multi/"
 

tests/models/test_nucleus_instance_segmentor.py

@@ -237,14 +237,14 @@ def test_cross_section_boundary_boxes() -> None:
     assert np.sum(flag - _flag) == 0, "Fail Cross Section Flag"
 
 
-def test_crash_segmentor(remote_sample: Callable, tmp_path: Path) -> None:
+def test_crash_segmentor(remote_sample: Callable, track_tmp_path: Path) -> None:
     """Test engine crash when given malformed input."""
-    root_save_dir = Path(tmp_path)
+    root_save_dir = Path(track_tmp_path)
     sample_wsi_svs = Path(remote_sample("svs-1-small"))
     sample_wsi_msk = remote_sample("small_svs_tissue_mask")
     sample_wsi_msk = np.load(sample_wsi_msk).astype(np.uint8)
-    imwrite(f"{tmp_path}/small_svs_tissue_mask.jpg", sample_wsi_msk)
-    sample_wsi_msk = tmp_path.joinpath("small_svs_tissue_mask.jpg")
+    imwrite(f"{track_tmp_path}/small_svs_tissue_mask.jpg", sample_wsi_msk)
+    sample_wsi_msk = track_tmp_path.joinpath("small_svs_tissue_mask.jpg")
 
     save_dir = f"{root_save_dir}/instance/"
 
@@ -285,17 +285,17 @@ def test_crash_segmentor(remote_sample: Callable, tmp_path: Path) -> None:
         )
 
 
-def test_functionality_ci(remote_sample: Callable, tmp_path: Path) -> None:
+def test_functionality_ci(remote_sample: Callable, track_tmp_path: Path) -> None:
     """Functionality test for nuclei instance segmentor."""
     gc.collect()
-    root_save_dir = Path(tmp_path)
+    root_save_dir = Path(track_tmp_path)
     mini_wsi_svs = Path(remote_sample("wsi4_512_512_svs"))
 
     resolution = 2.0
 
     reader = WSIReader.open(mini_wsi_svs)
     thumb = reader.slide_thumbnail(resolution=resolution, units="mpp")
-    mini_wsi_jpg = f"{tmp_path}/mini_svs.jpg"
+    mini_wsi_jpg = f"{track_tmp_path}/mini_svs.jpg"
     imwrite(mini_wsi_jpg, thumb)
 
     save_dir = f"{root_save_dir}/instance/"
@@ -335,11 +335,11 @@ def test_functionality_ci(remote_sample: Callable, tmp_path: Path) -> None:
 
 def test_functionality_merge_tile_predictions_ci(
     remote_sample: Callable,
-    tmp_path: Path,
+    track_tmp_path: Path,
 ) -> None:
     """Functional tests for merging tile predictions."""
     gc.collect()  # Force clean up everything on hold
-    save_dir = Path(f"{tmp_path}/output")
+    save_dir = Path(f"{track_tmp_path}/output")
     mini_wsi_svs = Path(remote_sample("wsi4_512_512_svs"))
 
     resolution = 0.5
@@ -438,10 +438,10 @@ def test_functionality_merge_tile_predictions_ci(
     toolbox_env.running_on_ci() or not ON_GPU,
     reason="Local test on machine with GPU.",
 )
-def test_functionality_local(remote_sample: Callable, tmp_path: Path) -> None:
+def test_functionality_local(remote_sample: Callable, track_tmp_path: Path) -> None:
     """Local functionality test for nuclei instance segmentor."""
-    root_save_dir = Path(tmp_path)
-    save_dir = Path(f"{tmp_path}/output")
+    root_save_dir = Path(track_tmp_path)
+    save_dir = Path(f"{track_tmp_path}/output")
     mini_wsi_svs = Path(remote_sample("wsi4_1k_1k_svs"))
 
     # * generate full output w/o parallel post-processing worker first
@@ -512,17 +512,17 @@ def test_functionality_local(remote_sample: Callable, tmp_path: Path) -> None:
 
 def test_cli_nucleus_instance_segment_ioconfig(
     remote_sample: Callable,
-    tmp_path: Path,
+    track_tmp_path: Path,
 ) -> None:
     """Test for nucleus segmentation with IOConfig."""
     mini_wsi_svs = Path(remote_sample("wsi4_512_512_svs"))
-    output_path = tmp_path / "output"
+    output_path = track_tmp_path / "output"
 
     resolution = 2.0
 
     reader = WSIReader.open(mini_wsi_svs)
     thumb = reader.slide_thumbnail(resolution=resolution, units="mpp")
-    mini_wsi_jpg = f"{tmp_path}/mini_svs.jpg"
+    mini_wsi_jpg = f"{track_tmp_path}/mini_svs.jpg"
     imwrite(mini_wsi_jpg, thumb)
 
     pretrained_weights = fetch_pretrained_weights("hovernet_fast-pannuke")
@@ -543,7 +543,7 @@ def test_cli_nucleus_instance_segment_ioconfig(
         "save_resolution": {"units": "mpp", "resolution": 8.0},
     }
 
-    with Path.open(tmp_path / "config.yaml", "w") as fptr:
+    with Path.open(track_tmp_path / "config.yaml", "w") as fptr:
         yaml.dump(config, fptr)
 
     runner = CliRunner()
@@ -564,7 +564,7 @@ def test_cli_nucleus_instance_segment_ioconfig(
             "--output-path",
             str(output_path),
             "--yaml-config-path",
-            str(tmp_path.joinpath("config.yaml")),
+            str(track_tmp_path.joinpath("config.yaml")),
         ],
     )
 
@@ -574,10 +574,12 @@ def test_cli_nucleus_instance_segment_ioconfig(
     assert output_path.joinpath("results.json").exists()
 
 
-def test_cli_nucleus_instance_segment(remote_sample: Callable, tmp_path: Path) -> None:
+def test_cli_nucleus_instance_segment(
+    remote_sample: Callable, track_tmp_path: Path
+) -> None:
     """Test for nucleus segmentation."""
     mini_wsi_svs = Path(remote_sample("wsi4_512_512_svs"))
-    output_path = tmp_path / "output"
+    output_path = track_tmp_path / "output"
 
     runner = CliRunner()
     nucleus_instance_segment_result = runner.invoke(