Simply ignore_warning functions

Author: rani-pinchuk

Lib/test/_test_multiprocessing.py

@@ -44,51 +44,32 @@ def check_syntax_warning(testcase, statement, errtext='',
     testcase.assertEqual(warns, [])
 
 
-def ignore_warnings(*, category):
+@contextlib.contextmanager
+def ignore_warnings(*, category, message=''):
     """Decorator to suppress warnings.
 
-    Use of context managers to hide warnings make diffs
-    more noisy and tools like 'git blame' less useful.
+    Can also be used as a context manager. This is not preferred,
+    because it makes diffs more noisy and tools like 'git blame' less useful.
+    But, it's useful for async functions.
     """
-    def decorator(test):
-        if inspect.iscoroutinefunction(test):
-            @functools.wraps(test)
-            async def async_wrapper(self, *args, **kwargs):
-                with warnings.catch_warnings():
-                    warnings.simplefilter('ignore', category=category)
-                    return await test(self, *args, **kwargs)
-            return async_wrapper
-        else:
-            @functools.wraps(test)
-            def wrapper(self, *args, **kwargs):
-                with warnings.catch_warnings():
-                    warnings.simplefilter('ignore', category=category)
-                    return test(self, *args, **kwargs)
-            return wrapper
-    return decorator
-
-
-def ignore_fork_in_thread_deprecation_warnings(test):
-    """Decorator to suppress the deprecation warnings related to running a fork within a thread.
+    with warnings.catch_warnings():
+        warnings.filterwarnings('ignore', category=category, message=message)
+        yield
+
+
+@contextlib.contextmanager
+def ignore_fork_in_thread_deprecation_warnings():
+    """Suppress deprecation warnings related to forking in multi-threaded code.
+
+    See gh-135427
+
+    Can be used as decorator (preferred) or context manager.
     """
-    if inspect.iscoroutinefunction(test):
-        @functools.wraps(test)
-        async def async_wrapper(self, *args, **kwargs):
-            with warnings.catch_warnings():
-                warnings.filterwarnings('ignore',
-                                        message=".*fork.*may lead to deadlocks in the child.*",
-                                        category=DeprecationWarning)
-                return await test(self, *args, **kwargs)
-        return async_wrapper
-    else:
-        @functools.wraps(test)
-        def wrapper(self, *args, **kwargs):
-            with warnings.catch_warnings():
-                warnings.filterwarnings('ignore',
-                                        message=".*fork.*may lead to deadlocks in the child.*",
-                                        category=DeprecationWarning)
-                return test(self, *args, **kwargs)
-        return wrapper
+    with ignore_warnings(
+        message=".*fork.*may lead to deadlocks in the child.*",
+        category=DeprecationWarning,
+    ):
+        yield
 
 
 class WarningsRecorder(object):

Lib/test/support/warnings_helper.py

@@ -1182,7 +1182,7 @@ async def runner():
 @support.requires_fork()
 class TestFork(unittest.IsolatedAsyncioTestCase):
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     async def test_fork_not_share_event_loop(self):
         # The forked process should not share the event loop with the parent
         loop = asyncio.get_running_loop()
@@ -1207,7 +1207,7 @@ async def test_fork_not_share_event_loop(self):
             self.assertEqual(result, b'NO LOOP')
             wait_process(pid, exitcode=0)
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     @hashlib_helper.requires_hashdigest('md5')
     @support.skip_if_sanitizer("TSAN doesn't support threads after fork", thread=True)
     def test_fork_signal_handling(self):
@@ -1255,7 +1255,7 @@ async def func():
         self.assertFalse(parent_handled.is_set())
         self.assertTrue(child_handled.is_set())
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     @hashlib_helper.requires_hashdigest('md5')
     @support.skip_if_sanitizer("TSAN doesn't support threads after fork", thread=True)
     def test_fork_asyncio_run(self):
@@ -1276,7 +1276,7 @@ async def child_main():
 
         self.assertEqual(result.value, 42)
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     @hashlib_helper.requires_hashdigest('md5')
     @support.skip_if_sanitizer("TSAN doesn't support threads after fork", thread=True)
     def test_fork_asyncio_subprocess(self):

Lib/test/test_asyncio/test_unix_events.py

@@ -2546,7 +2546,7 @@ def run_child(self, child, terminal_input):
         finally:
             signal.signal(signal.SIGHUP, old_sighup)
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def _run_child(self, child, terminal_input):
         r, w = os.pipe()  # Pipe test results from child back to parent
         try:

Lib/test/test_builtin.py

@@ -43,12 +43,12 @@ class ExecutorTest:
 
     # Executor.shutdown() and context manager usage is tested by
     # ExecutorShutdownTest.
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def test_submit(self):
         future = self.executor.submit(pow, 2, 8)
         self.assertEqual(256, future.result())
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def test_submit_keyword(self):
         future = self.executor.submit(mul, 2, y=8)
         self.assertEqual(16, future.result())
@@ -59,7 +59,7 @@ def test_submit_keyword(self):
         with self.assertRaises(TypeError):
             self.executor.submit(arg=1)
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def test_map(self):
         self.assertEqual(
                 list(self.executor.map(pow, range(10), range(10))),
@@ -69,15 +69,15 @@ def test_map(self):
                 list(self.executor.map(pow, range(10), range(10), chunksize=3)),
                 list(map(pow, range(10), range(10))))
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def test_map_exception(self):
         i = self.executor.map(divmod, [1, 1, 1, 1], [2, 3, 0, 5])
         self.assertEqual(i.__next__(), (0, 1))
         self.assertEqual(i.__next__(), (0, 1))
         with self.assertRaises(ZeroDivisionError):
             i.__next__()
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     @support.requires_resource('walltime')
     def test_map_timeout(self):
         results = []
@@ -113,30 +113,30 @@ def test_map_buffersize_value_validation(self):
                 ):
                     self.executor.map(str, range(4), buffersize=buffersize)
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def test_map_buffersize(self):
         ints = range(4)
         for buffersize in (1, 2, len(ints), len(ints) * 2):
             with self.subTest(buffersize=buffersize):
                 res = self.executor.map(str, ints, buffersize=buffersize)
                 self.assertListEqual(list(res), ["0", "1", "2", "3"])
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def test_map_buffersize_on_multiple_iterables(self):
         ints = range(4)
         for buffersize in (1, 2, len(ints), len(ints) * 2):
             with self.subTest(buffersize=buffersize):
                 res = self.executor.map(add, ints, ints, buffersize=buffersize)
                 self.assertListEqual(list(res), [0, 2, 4, 6])
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def test_map_buffersize_on_infinite_iterable(self):
         res = self.executor.map(str, itertools.count(), buffersize=2)
         self.assertEqual(next(res, None), "0")
         self.assertEqual(next(res, None), "1")
         self.assertEqual(next(res, None), "2")
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def test_map_buffersize_on_multiple_infinite_iterables(self):
         res = self.executor.map(
             add,
@@ -156,7 +156,7 @@ def test_map_buffersize_without_iterable(self):
         res = self.executor.map(str, buffersize=2)
         self.assertIsNone(next(res, None))
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def test_map_buffersize_when_buffer_is_full(self):
         ints = iter(range(4))
         buffersize = 2
@@ -168,15 +168,15 @@ def test_map_buffersize_when_buffer_is_full(self):
             msg="should have fetched only `buffersize` elements from `ints`.",
         )
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def test_shutdown_race_issue12456(self):
         # Issue #12456: race condition at shutdown where trying to post a
         # sentinel in the call queue blocks (the queue is full while processes
         # have exited).
         self.executor.map(str, [2] * (self.worker_count + 1))
         self.executor.shutdown()
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     @support.cpython_only
     def test_no_stale_references(self):
         # Issue #16284: check that the executors don't unnecessarily hang onto
@@ -221,7 +221,7 @@ def test_max_workers_negative(self):
                                         "than 0"):
                 self.executor_type(max_workers=number)
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def test_free_reference(self):
         # Issue #14406: Result iterator should not keep an internal
         # reference to result objects.
@@ -234,7 +234,7 @@ def test_free_reference(self):
                 if wr() is None:
                     break
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def test_swallows_falsey_exceptions(self):
         # see gh-132063: Prevent exceptions that evaluate as falsey
         # from being ignored.

Lib/test/test_concurrent_futures/executor.py

@@ -20,7 +20,7 @@ def mul(x, y):
 
 
 class AsCompletedTests:
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def test_no_timeout(self):
         future1 = self.executor.submit(mul, 2, 21)
         future2 = self.executor.submit(mul, 7, 6)
@@ -37,7 +37,7 @@ def test_no_timeout(self):
                  future1, future2]),
                 completed)
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def test_future_times_out(self):
         """Test ``futures.as_completed`` timing out before
         completing it's final future."""
@@ -65,7 +65,7 @@ def test_future_times_out(self):
                 # Check that ``future`` wasn't completed.
                 self.assertEqual(completed_futures, already_completed)
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def test_duplicate_futures(self):
         # Issue 20367. Duplicate futures should not raise exceptions or give
         # duplicate responses.

Lib/test/test_concurrent_futures/test_as_completed.py

@@ -112,7 +112,7 @@ def _fail_on_deadlock(self, executor):
         print(f"\nTraceback:\n {tb}", file=sys.__stderr__)
         self.fail(f"Executor deadlock:\n\n{tb}")
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def _check_error(self, error, func, *args, ignore_stderr=False):
         # test for deadlock caused by crashes or exiting in a pool
         self.executor.shutdown(wait=True)
@@ -201,7 +201,7 @@ def test_exit_during_result_unpickle_in_result_handler(self):
         # the result_handler thread
         self._check_error(BrokenProcessPool, _return_instance, ExitAtUnpickle)
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     @support.skip_if_sanitizer("UBSan: explicit SIGSEV not allowed", ub=True)
     def test_shutdown_deadlock(self):
         # Test that the pool calling shutdown do not cause deadlock
@@ -215,7 +215,7 @@ def test_shutdown_deadlock(self):
             with self.assertRaises(BrokenProcessPool):
                 f.result()
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def test_shutdown_deadlock_pickle(self):
         # Test that the pool calling shutdown with wait=False does not cause
         # a deadlock if a task fails at pickle after the shutdown call.
@@ -242,7 +242,7 @@ def test_shutdown_deadlock_pickle(self):
         # dangling threads
         executor_manager.join()
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     @support.skip_if_sanitizer("UBSan: explicit SIGSEV not allowed", ub=True)
     def test_crash_big_data(self):
         # Test that there is a clean exception instead of a deadlock when a
@@ -259,7 +259,7 @@ def test_crash_big_data(self):
 
         executor.shutdown(wait=True)
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def test_gh105829_should_not_deadlock_if_wakeup_pipe_full(self):
         # Issue #105829: The _ExecutorManagerThread wakeup pipe could
         # fill up and block. See: https://github.com/python/cpython/issues/105829

Lib/test/test_concurrent_futures/test_deadlock.py

@@ -49,7 +49,7 @@ def setUp(self):
                                     initargs=('initialized',))
         super().setUp()
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def test_initializer(self):
         futures = [self.executor.submit(get_init_status)
                    for _ in range(self.worker_count)]
@@ -76,7 +76,7 @@ def setUp(self):
             self.executor_kwargs = dict(initializer=init_fail)
         super().setUp()
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def test_initializer(self):
         with self._assert_logged('ValueError: error in initializer'):
             try:

Lib/test/test_concurrent_futures/test_init.py

@@ -49,7 +49,7 @@ def test_max_workers_too_large(self):
                                     "max_workers must be <= 61"):
             futures.ProcessPoolExecutor(max_workers=62)
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def test_killed_child(self):
         # When a child process is abruptly terminated, the whole pool gets
         # "broken".
@@ -62,7 +62,7 @@ def test_killed_child(self):
         # Submitting other jobs fails as well.
         self.assertRaises(BrokenProcessPool, self.executor.submit, pow, 2, 8)
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def test_map_chunksize(self):
         def bad_map():
             list(self.executor.map(pow, range(40), range(40), chunksize=-1))
@@ -83,7 +83,7 @@ def bad_map():
     def _test_traceback(cls):
         raise RuntimeError(123) # some comment
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def test_traceback(self):
         # We want ensure that the traceback from the child process is
         # contained in the traceback raised in the main process.
@@ -106,7 +106,7 @@ def test_traceback(self):
         self.assertIn('raise RuntimeError(123) # some comment',
                       f1.getvalue())
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     @hashlib_helper.requires_hashdigest('md5')
     def test_ressources_gced_in_workers(self):
         # Ensure that argument for a job are correctly gc-ed after the job
@@ -127,7 +127,7 @@ def test_ressources_gced_in_workers(self):
         mgr.shutdown()
         mgr.join()
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def test_saturation(self):
         executor = self.executor
         mp_context = self.get_context()
@@ -213,7 +213,7 @@ def test_max_tasks_early_shutdown(self):
         for i, future in enumerate(futures):
             self.assertEqual(future.result(), mul(i, i))
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def test_python_finalization_error(self):
         # gh-109047: Catch RuntimeError on thread creation
         # during Python finalization.
@@ -264,7 +264,7 @@ def test_force_shutdown_workers_invalid_op(self):
                               executor._force_shutdown,
                               operation='invalid operation'),
 
-    @warnings_helper.ignore_fork_in_thread_deprecation_warnings  # gh-135427
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     @parameterize(*FORCE_SHUTDOWN_PARAMS)
     def test_force_shutdown_workers(self, function_name):
         manager = self.get_context().Manager()