Fix warning assertions for pytest 8 compatibility

Author: AKuederle

tests/conftest.py

@@ -1,8 +1,23 @@
+import warnings
+from contextlib import contextmanager
 from typing import Any
 
+import pytest
+
 from tpcp import BaseTpcpObject
 from tpcp._base import _BaseTpcpObject
 
 
 def _get_params_without_nested_class(instance: BaseTpcpObject) -> dict[str, Any]:
     return {k: v for k, v in instance.get_params().items() if not isinstance(v, _BaseTpcpObject)}
+
+
+@contextmanager
+def warns_or_none(expected_warning):
+    if expected_warning is None:
+        with warnings.catch_warnings(record=True) as caught:
+            warnings.simplefilter("always")
+            yield caught
+    else:
+        with pytest.warns(expected_warning) as caught:
+            yield caught

tests/test_caching.py

@@ -106,7 +106,8 @@ def test_caching_twice_same_instance(self, example_class):
             getattr(example, action_name)(2)
         assert example.result_1_ == 5 * multiplier
 
-        with pytest.warns(None) as w:
+        with warnings.catch_warnings(record=True) as w:
+            warnings.simplefilter("always")
             getattr(example, action_name)(3)
         assert example.result_1_ == 6 * multiplier
         assert not w
@@ -126,7 +127,8 @@ def test_caching_twice_new_instance(self, example_class):
         assert example.result_1_ == 5 * multiplier
 
         example = example_class(1, 2)
-        with pytest.warns(None) as w:
+        with warnings.catch_warnings(record=True) as w:
+            warnings.simplefilter("always")
             getattr(example, action_name)(3)
         assert example.result_1_ == 6 * multiplier
         assert not w
@@ -165,7 +167,8 @@ def test_cache_only(self, example_class):
         example = example.clone()
 
         # Now in the cached version
-        with pytest.warns(None) as w:
+        with warnings.catch_warnings(record=True) as w:
+            warnings.simplefilter("always")
             getattr(example, action_name)(2)
         assert not w
         assert example.result_1_ == 5 * multiplier
@@ -197,7 +200,8 @@ def worker_func(pipe):
             if self.cache_method_name == "disk" and restore_in_parallel_process is True:
                 # Disk cache can work across processes. This means, already on the first call in the new process,
                 # we should get the cached result.
-                with pytest.warns(None) as w:
+                with warnings.catch_warnings(record=True) as w:
+                    warnings.simplefilter("always")
                     pipe.action(1)
                 assert not w
             else:
@@ -207,7 +211,8 @@ def worker_func(pipe):
 
             if restore_in_parallel_process is True:
                 # Id we set the restore option to True, the second call should be correctly cached
-                with pytest.warns(None) as w:
+                with warnings.catch_warnings(record=True) as w:
+                    warnings.simplefilter("always")
                     pipe.action(1)
                 assert not w
             else:
@@ -255,7 +260,8 @@ def test_joblib_only(self, joblib_cache, hybrid_cache_clear):
 
         assert r == 3
 
-        with pytest.warns(None) as w:
+        with warnings.catch_warnings(record=True) as w:
+            warnings.simplefilter("always")
             r = cached_func(1, 2)
 
         assert r == 3
@@ -269,7 +275,8 @@ def test_lru_only(self, hybrid_cache_clear):
 
         assert r == 3
 
-        with pytest.warns(None) as w:
+        with warnings.catch_warnings(record=True) as w:
+            warnings.simplefilter("always")
             r = cached_func(1, 2)
 
         assert r == 3
@@ -288,7 +295,8 @@ def test_staggered_cache(self, joblib_cache_verbose, hybrid_cache_clear, capfd):
 
         assert r == 3
 
-        with pytest.warns(None) as w:
+        with warnings.catch_warnings(record=True) as w:
+            warnings.simplefilter("always")
             r = cached_func(1, 2)
 
         # This should not hit the joblib cache, as the lru cache should have been used
@@ -317,7 +325,8 @@ def test_joblib_cache_survives_clear(self, joblib_cache_verbose, hybrid_cache_cl
 
         cached_func_new = hybrid_cache(joblib_cache_verbose, 2)(example_func)
 
-        with pytest.warns(None) as w:
+        with warnings.catch_warnings(record=True) as w:
+            warnings.simplefilter("always")
             r = cached_func_new(1, 2)
 
         # This time this should hit the joblib cache, as the lru cache should have been cleared
@@ -330,7 +339,8 @@ def test_joblib_cache_survives_clear(self, joblib_cache_verbose, hybrid_cache_cl
         assert not w
 
         # And now the lru cache should be used again
-        with pytest.warns(None) as w:
+        with warnings.catch_warnings(record=True) as w:
+            warnings.simplefilter("always")
             r = cached_func_new(1, 2)
 
         # This time this should hit the joblib cache, as the lru cache should have been cleared

tests/test_pipelines/test_optimize.py

@@ -1,3 +1,4 @@
+import warnings
 from tempfile import TemporaryDirectory
 from typing import Union
 from unittest.mock import patch
@@ -8,6 +9,7 @@
 import pytest
 from sklearn.model_selection import ParameterGrid, PredefinedSplit
 
+from tests.conftest import warns_or_none
 from tests.test_pipelines.conftest import (
     DummyDataset,
     DummyOptimizablePipeline,
@@ -147,7 +149,7 @@ def test_return_optimized_single(self, return_optimized):
         if isinstance(return_optimized, str) and not return_optimized.endswith("score"):
             warning = UserWarning
 
-        with pytest.warns(warning) as w:
+        with warns_or_none(warning) as w:
             gs.optimize(DummyDataset())
 
         if isinstance(return_optimized, str) and not return_optimized.endswith("score"):
@@ -760,7 +762,7 @@ def test_safe_optimize(self, use_safe):
         with patch.object(DummyOptimizablePipelineUnsafe, "self_optimize", return_value=optimized_pipe) as mock:
             mock.__name__ = "self_optimize"
             warning = PotentialUserErrorWarning if use_safe else None
-            with pytest.warns(warning) as w:
+            with warns_or_none(warning) as w:
                 Optimize(DummyOptimizablePipelineUnsafe(), safe_optimize=use_safe).optimize(ds)
 
             if use_safe:
@@ -818,16 +820,18 @@ def test_warning(self):
 
         assert len(w.list) == 1
 
-        with pytest.warns(None) as w:
+        with warnings.catch_warnings(record=True) as w:
+            warnings.simplefilter("always")
             DummyOptimize(DummyPipeline()).optimize(dataset=None)
 
-        assert len(w.list) == 0
+        assert len(w) == 0
 
     def test_warning_suppression(self):
-        with pytest.warns(None) as w:
+        with warnings.catch_warnings(record=True) as w:
+            warnings.simplefilter("always")
             DummyOptimize(DummyOptimizablePipeline(), ignore_potential_user_error_warning=True).optimize(dataset=None)
 
-        assert len(w.list) == 0
+        assert len(w) == 0
 
 
 class TestOptimizeBase:

tests/test_safe_decorator.py

@@ -3,6 +3,7 @@
 
 import pytest
 
+from tests.conftest import warns_or_none
 from tests.test_pipelines.conftest import DummyDataset
 from tpcp import (
     Dataset,
@@ -90,7 +91,7 @@ def test_wrapper_checks_name(self, name, warn):
         ds = DummyDataset()
 
         warning = PotentialUserErrorWarning if warn else None
-        with pytest.warns(warning) as w:
+        with warns_or_none(warning) as w:
             make_action_safe(test_func)(DummyActionPipelineUnsafe(), ds)
 
         if warn:
@@ -148,7 +149,7 @@ def test_wrapper_checks_name(self, name, warn):
         ds = DummyDataset()
 
         warning = PotentialUserErrorWarning if warn else None
-        with pytest.warns(warning) as w:
+        with warns_or_none(warning) as w:
             make_optimize_safe(test_func)(DummyOptimizablePipelineUnsafe(), ds)
 
         if warn:
@@ -166,7 +167,7 @@ def test_optimize_warns(self, output, warn):
                 DummyOptimizablePipelineUnsafe.self_optimize
             )
             warning = PotentialUserErrorWarning if warn else None
-            with pytest.warns(warning) as w:
+            with warns_or_none(warning) as w:
                 DummyOptimizablePipelineUnsafe().self_optimize(ds)
 
             if len(w) > 0: