pythongh-108416: Mark slow test methods with @requires_resource('cpu') (

pythonGH-108421)

Only mark tests which spend significant system or user time,
by itself or in subprocesses.

Lib/test/_test_multiprocessing.py

@@ -329,6 +329,7 @@ def test_set_executable(self):
             p.join()
             self.assertEqual(p.exitcode, 0)
 
+    @support.requires_resource('cpu')
     def test_args_argument(self):
         # bpo-45735: Using list or tuple as *args* in constructor could
         # achieve the same effect.
@@ -4504,6 +4505,7 @@ def test_finalize(self):
         result = [obj for obj in iter(conn.recv, 'STOP')]
         self.assertEqual(result, ['a', 'b', 'd10', 'd03', 'd02', 'd01', 'e'])
 
+    @support.requires_resource('cpu')
     def test_thread_safety(self):
         # bpo-24484: _run_finalizers() should be thread-safe
         def cb():

Lib/test/pickletester.py

@@ -2576,6 +2576,7 @@ def check_frame_opcodes(self, pickled):
             self.assertLess(pos - frameless_start, self.FRAME_SIZE_MIN)
 
     @support.skip_if_pgo_task
+    @support.requires_resource('cpu')
     def test_framing_many_objects(self):
         obj = list(range(10**5))
         for proto in range(4, pickle.HIGHEST_PROTOCOL + 1):

Lib/test/test_ast.py

@@ -2031,6 +2031,7 @@ def test_nameconstant(self):
             'ast.NameConstant is deprecated and will be removed in Python 3.14; use ast.Constant instead',
         ])
 
+    @support.requires_resource('cpu')
     def test_stdlib_validates(self):
         stdlib = os.path.dirname(ast.__file__)
         tests = [fn for fn in os.listdir(stdlib) if fn.endswith(".py")]

Lib/test/test_buffer.py

@@ -1029,6 +1029,7 @@ def match(req, flag):
                     ndim=ndim, shape=shape, strides=strides,
                     lst=lst, sliced=sliced)
 
+    @support.requires_resource('cpu')
     def test_ndarray_getbuf(self):
         requests = (
             # distinct flags
@@ -2760,6 +2761,7 @@ def iter_roundtrip(ex, m, items, fmt):
             m = memoryview(ex)
             iter_roundtrip(ex, m, items, fmt)
 
+    @support.requires_resource('cpu')
     def test_memoryview_cast_1D_ND(self):
         # Cast between C-contiguous buffers. At least one buffer must
         # be 1D, at least one format must be 'c', 'b' or 'B'.

Lib/test/test_builtin.py

@@ -952,6 +952,7 @@ def test_filter_pickle(self):
             f2 = filter(filter_char, "abcdeabcde")
             self.check_iter_pickle(f1, list(f2), proto)
 
+    @support.requires_resource('cpu')
     def test_filter_dealloc(self):
         # Tests recursive deallocation of nested filter objects using the
         # thrashcan mechanism. See gh-102356 for more details.

Lib/test/test_compile.py

@@ -784,6 +784,7 @@ def test_path_like_objects(self):
         # An implicit test for PyUnicode_FSDecoder().
         compile("42", FakePath("test_compile_pathlike"), "single")
 
+    @support.requires_resource('cpu')
     def test_stack_overflow(self):
         # bpo-31113: Stack overflow when compile a long sequence of
         # complex statements.

Lib/test/test_compileall.py

@@ -551,6 +551,7 @@ def test_no_args_compiles_path(self):
             self.assertNotCompiled(self.barfn)
 
     @without_source_date_epoch  # timestamp invalidation test
+    @support.requires_resource('cpu')
     def test_no_args_respects_force_flag(self):
         bazfn = script_helper.make_script(self.directory, 'baz', '')
         with self.temporary_pycache_prefix() as env:
@@ -568,6 +569,7 @@ def test_no_args_respects_force_flag(self):
         mtime2 = os.stat(pycpath).st_mtime
         self.assertNotEqual(mtime, mtime2)
 
+    @support.requires_resource('cpu')
     def test_no_args_respects_quiet_flag(self):
         script_helper.make_script(self.directory, 'baz', '')
         with self.temporary_pycache_prefix() as env:

Lib/test/test_concurrent_futures/test_thread_pool.py

@@ -6,6 +6,7 @@
 import threading
 import unittest
 from concurrent import futures
+from test import support
 
 from .executor import ExecutorTest, mul
 from .util import BaseTestCase, ThreadPoolMixin, setup_module
@@ -49,6 +50,7 @@ def test_idle_thread_reuse(self):
         executor.shutdown(wait=True)
 
     @unittest.skipUnless(hasattr(os, 'register_at_fork'), 'need os.register_at_fork')
+    @support.requires_resource('cpu')
     def test_hang_global_shutdown_lock(self):
         # bpo-45021: _global_shutdown_lock should be reinitialized in the child
         # process, otherwise it will never exit

Lib/test/test_context.py

@@ -6,6 +6,7 @@
 import time
 import unittest
 import weakref
+from test import support
 from test.support import threading_helper
 
 try:
@@ -570,6 +571,7 @@ def test_hamt_collision_3(self):
 
         self.assertEqual({k.name for k in h.keys()}, {'C', 'D', 'E'})
 
+    @support.requires_resource('cpu')
     def test_hamt_stress(self):
         COLLECTION_SIZE = 7000
         TEST_ITERS_EVERY = 647

Lib/test/test_cppext/__init__.py

@@ -16,9 +16,11 @@
 
 @support.requires_subprocess()
 class TestCPPExt(unittest.TestCase):
+    @support.requires_resource('cpu')
     def test_build_cpp11(self):
         self.check_build(False, '_testcpp11ext')
 
+    @support.requires_resource('cpu')
     def test_build_cpp03(self):
         self.check_build(True, '_testcpp03ext')
 

Lib/test/test_descr.py

@@ -4457,6 +4457,7 @@ class Oops(object):
         o.whatever = Provoker(o)
         del o
 
+    @support.requires_resource('cpu')
     def test_wrapper_segfault(self):
         # SF 927248: deeply nested wrappers could cause stack overflow
         f = lambda:None

Lib/test/test_email/test_email.py

@@ -38,6 +38,7 @@
 from email import quoprimime
 from email import utils
 
+from test import support
 from test.support import threading_helper
 from test.support.os_helper import unlink
 from test.test_email import openfile, TestEmailBase
@@ -3358,6 +3359,7 @@ def test_getaddresses_header_obj(self):
         self.assertEqual(addrs[0][1], '[email protected]')
 
     @threading_helper.requires_working_threading()
+    @support.requires_resource('cpu')
     def test_make_msgid_collisions(self):
         # Test make_msgid uniqueness, even with multiple threads
         class MsgidsThread(Thread):

Lib/test/test_exceptions.py

@@ -1350,6 +1350,7 @@ def g():
 
 
     @cpython_only
+    @support.requires_resource('cpu')
     def test_trashcan_recursion(self):
         # See bpo-33930
 

Lib/test/test_gdb.py

@@ -317,6 +317,7 @@ def assertGdbRepr(self, val, exp_repr=None):
                          ('%r did not equal expected %r; full output was:\n%s'
                           % (gdb_repr, exp_repr, gdb_output)))
 
+    @support.requires_resource('cpu')
     def test_int(self):
         'Verify the pretty-printing of various int values'
         self.assertGdbRepr(42)
@@ -343,6 +344,7 @@ def test_lists(self):
         self.assertGdbRepr([])
         self.assertGdbRepr(list(range(5)))
 
+    @support.requires_resource('cpu')
     def test_bytes(self):
         'Verify the pretty-printing of bytes'
         self.assertGdbRepr(b'')
@@ -357,6 +359,7 @@ def test_bytes(self):
 
         self.assertGdbRepr(bytes([b for b in range(255)]))
 
+    @support.requires_resource('cpu')
     def test_strings(self):
         'Verify the pretty-printing of unicode strings'
         # We cannot simply call locale.getpreferredencoding() here,
@@ -407,6 +410,7 @@ def test_tuples(self):
         self.assertGdbRepr((1,), '(1,)')
         self.assertGdbRepr(('foo', 'bar', 'baz'))
 
+    @support.requires_resource('cpu')
     def test_sets(self):
         'Verify the pretty-printing of sets'
         if (gdb_major_version, gdb_minor_version) < (7, 3):
@@ -425,6 +429,7 @@ def test_sets(self):
 id(s)''')
         self.assertEqual(gdb_repr, "{'b'}")
 
+    @support.requires_resource('cpu')
     def test_frozensets(self):
         'Verify the pretty-printing of frozensets'
         if (gdb_major_version, gdb_minor_version) < (7, 3):
@@ -828,6 +833,7 @@ def test_bt_full(self):
 
     @unittest.skipIf(python_is_optimized(),
                      "Python was compiled with optimizations")
+    @support.requires_resource('cpu')
     def test_threads(self):
         'Verify that "py-bt" indicates threads that are waiting for the GIL'
         cmd = '''
@@ -889,6 +895,7 @@ def test_gc(self):
 
     @unittest.skipIf(python_is_optimized(),
                      "Python was compiled with optimizations")
+    @support.requires_resource('cpu')
     # Some older versions of gdb will fail with
     #  "Cannot find new threads: generic error"
     # unless we add LD_PRELOAD=PATH-TO-libpthread.so.1 as a workaround

Lib/test/test_interpreters.py

@@ -469,12 +469,14 @@ class StressTests(TestBase):
     # In these tests we generally want a lot of interpreters,
     # but not so many that any test takes too long.
 
+    @support.requires_resource('cpu')
     def test_create_many_sequential(self):
         alive = []
         for _ in range(100):
             interp = interpreters.create()
             alive.append(interp)
 
+    @support.requires_resource('cpu')
     def test_create_many_threaded(self):
         alive = []
         def task():

Lib/test/test_itertools.py

@@ -2401,6 +2401,7 @@ def gen2(x):
         self.assertEqual(hist, [0,1])
 
     @support.skip_if_pgo_task
+    @support.requires_resource('cpu')
     def test_long_chain_of_empty_iterables(self):
         # Make sure itertools.chain doesn't run into recursion limits when
         # dealing with long chains of empty iterables. Even with a high

Lib/test/test_largefile.py

@@ -8,7 +8,7 @@
 import socket
 import shutil
 import threading
-from test.support import requires, bigmemtest
+from test.support import requires, bigmemtest, requires_resource
 from test.support import SHORT_TIMEOUT
 from test.support import socket_helper
 from test.support.os_helper import TESTFN, unlink
@@ -173,6 +173,7 @@ class TestCopyfile(LargeFileTest, unittest.TestCase):
     # Exact required disk space would be (size * 2), but let's give it a
     # bit more tolerance.
     @skip_no_disk_space(TESTFN, size * 2.5)
+    @requires_resource('cpu')
     def test_it(self):
         # Internally shutil.copyfile() can use "fast copy" methods like
         # os.sendfile().
@@ -222,6 +223,7 @@ def run(sock):
     # Exact required disk space would be (size * 2), but let's give it a
     # bit more tolerance.
     @skip_no_disk_space(TESTFN, size * 2.5)
+    @requires_resource('cpu')
     def test_it(self):
         port = socket_helper.find_unused_port()
         with socket.create_server(("", port)) as sock:

Lib/test/test_multibytecodec.py

@@ -363,6 +363,7 @@ def test_iso2022_jp_g0(self):
             e = '\u3406'.encode(encoding)
             self.assertFalse(any(x > 0x80 for x in e))
 
+    @support.requires_resource('cpu')
     def test_bug1572832(self):
         for x in range(0x10000, 0x110000):
             # Any ISO 2022 codec will cause the segfault

Lib/test/test_peepholer.py

@@ -4,6 +4,7 @@
 import textwrap
 import unittest
 
+from test import support
 from test.support.bytecode_helper import BytecodeTestCase, CfgOptimizationTestCase
 
 
@@ -522,6 +523,7 @@ def genexpr():
             return (y for x in a for y in [f(x)])
         self.assertEqual(count_instr_recursively(genexpr, 'FOR_ITER'), 1)
 
+    @support.requires_resource('cpu')
     def test_format_combinations(self):
         flags = '-+ #0'
         testcases = [

Lib/test/test_runpy.py

@@ -12,7 +12,7 @@
 import textwrap
 import unittest
 import warnings
-from test.support import no_tracing, verbose, requires_subprocess
+from test.support import no_tracing, verbose, requires_subprocess, requires_resource
 from test.support.import_helper import forget, make_legacy_pyc, unload
 from test.support.os_helper import create_empty_file, temp_dir
 from test.support.script_helper import make_script, make_zip_script
@@ -733,6 +733,7 @@ def test_zipfile_error(self):
             self._check_import_error(zip_name, msg)
 
     @no_tracing
+    @requires_resource('cpu')
     def test_main_recursion_error(self):
         with temp_dir() as script_dir, temp_dir() as dummy_dir:
             mod_name = '__main__'

Lib/test/test_selectors.py

@@ -455,6 +455,7 @@ class ScalableSelectorMixIn:
     # see issue #18963 for why it's skipped on older OS X versions
     @support.requires_mac_ver(10, 5)
     @unittest.skipUnless(resource, "Test needs resource module")
+    @support.requires_resource('cpu')
     def test_above_fd_setsize(self):
         # A scalable implementation should have no problem with more than
         # FD_SETSIZE file descriptors. Since we don't know the value, we just

Lib/test/test_source_encoding.py

@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 
 import unittest
-from test.support import script_helper, captured_stdout, requires_subprocess
+from test.support import script_helper, captured_stdout, requires_subprocess, requires_resource
 from test.support.os_helper import TESTFN, unlink, rmtree
 from test.support.import_helper import unload
 import importlib
@@ -250,6 +250,7 @@ def test_crcrcrlf2(self):
 class UTF8ValidatorTest(unittest.TestCase):
     @unittest.skipIf(not sys.platform.startswith("linux"),
                      "Too slow to run on non-Linux platforms")
+    @requires_resource('cpu')
     def test_invalid_utf8(self):
         # This is a port of test_utf8_decode_invalid_sequences in
         # test_unicode.py to exercise the separate utf8 validator in

Lib/test/test_statistics.py

@@ -2145,6 +2145,7 @@ def test_integer_sqrt_of_frac_rto(self):
             self.assertTrue(m * (r - 1)**2 < n < m * (r + 1)**2)
 
     @requires_IEEE_754
+    @support.requires_resource('cpu')
     def test_float_sqrt_of_frac(self):
 
         def is_root_correctly_rounded(x: Fraction, root: float) -> bool:
@@ -2849,6 +2850,7 @@ def test_cdf(self):
         self.assertTrue(math.isnan(X.cdf(float('NaN'))))
 
     @support.skip_if_pgo_task
+    @support.requires_resource('cpu')
     def test_inv_cdf(self):
         NormalDist = self.module.NormalDist
 

Lib/test/test_subprocess.py

@@ -1292,6 +1292,7 @@ def test_bufsize_equal_one_binary_mode(self):
         with self.assertWarnsRegex(RuntimeWarning, 'line buffering'):
             self._test_bufsize_equal_one(line, b'', universal_newlines=False)
 
+    @support.requires_resource('cpu')
     def test_leaking_fds_on_error(self):
         # see bug #5179: Popen leaks file descriptors to PIPEs if
         # the child fails to execute; this will eventually exhaust

Lib/test/test_support.py

@@ -497,6 +497,7 @@ def check_options(self, args, func, expected=None):
         self.assertEqual(proc.stdout.rstrip(), repr(expected))
         self.assertEqual(proc.returncode, 0)
 
+    @support.requires_resource('cpu')
     def test_args_from_interpreter_flags(self):
         # Test test.support.args_from_interpreter_flags()
         for opts in (

Lib/test/test_sys_settrace.py

@@ -2834,6 +2834,7 @@ def test_jump_extended_args_unpack_ex_tricky(output):
         ) = output.append(4) or "Spam"
         output.append(5)
 
+    @support.requires_resource('cpu')
     def test_jump_extended_args_for_iter(self):
         # In addition to failing when extended arg handling is broken, this can
         # also hang for a *very* long time:

Lib/test/test_tools/test_freeze.py

@@ -17,10 +17,8 @@
 @support.skip_if_buildbot('not all buildbots have enough space')
 class TestFreeze(unittest.TestCase):
 
+    @support.requires_resource('cpu') # Building Python is slow
     def test_freeze_simple_script(self):
-        # Building Python is slow
-        support.requires('cpu')
-
         script = textwrap.dedent("""
             import sys
             print('running...')

Lib/test/test_trace.py

@@ -1,7 +1,7 @@
 import os
 from pickle import dump
 import sys
-from test.support import captured_stdout
+from test.support import captured_stdout, requires_resource
 from test.support.os_helper import (TESTFN, rmtree, unlink)
 from test.support.script_helper import assert_python_ok, assert_python_failure
 import textwrap
@@ -367,6 +367,7 @@ def _coverage(self, tracer,
         r = tracer.results()
         r.write_results(show_missing=True, summary=True, coverdir=TESTFN)
 
+    @requires_resource('cpu')
     def test_coverage(self):
         tracer = trace.Trace(trace=0, count=1)
         with captured_stdout() as stdout: