mirrors/cpython
mirrors/cpython
0
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2024-11-21 12:59:38 +01:00
Code Issues Releases Activity

gh-108416: Mark slow but not CPU bound test methods with requires_resource('walltime') (GH-108480)

Browse Source
This commit is contained in:
Serhiy Storchaka 2023-09-05 17:56:30 +03:00 committed by GitHub
parent f980cc19b9
commit 1e0d62793a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
18 changed files with 48 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
Lib/test/_test_multiprocessing.py
View File

@ -675,6 +675,7 @@ class _TestProcess(BaseTestCase):
close_queue(q)
@support.requires_resource('walltime')
def test_many_processes(self):
if self.TYPE == 'threads':
self.skipTest('test not appropriate for {}'.format(self.TYPE))
@ -4991,6 +4992,7 @@ class TestWait(unittest.TestCase):
def test_wait_socket_slow(self):
self.test_wait_socket(True)
@support.requires_resource('walltime')
def test_wait_timeout(self):
from multiprocessing.connection import wait
@ -5019,6 +5021,7 @@ class TestWait(unittest.TestCase):
sem.release()
time.sleep(period)
@support.requires_resource('walltime')
def test_wait_integer(self):
from multiprocessing.connection import wait

4
Lib/test/libregrtest/cmdline.py
View File

@ -107,6 +107,8 @@ resources to test. Currently only the following are defined:
cpu - Used for certain CPU-heavy tests.
walltime - Long running but not CPU-bound tests.
subprocess Run all tests for the subprocess module.
urlfetch - It is okay to download files required on testing.
@ -129,7 +131,7 @@ Pattern examples:
ALL_RESOURCES = ('audio', 'curses', 'largefile', 'network',
'decimal', 'cpu', 'subprocess', 'urlfetch', 'gui')
'decimal', 'cpu', 'subprocess', 'urlfetch', 'gui', 'walltime')
# Other resources excluded from --use=all:
#

1
Lib/test/test_concurrent_futures/executor.py
View File

@ -53,6 +53,7 @@ class ExecutorTest:
self.assertEqual(i.__next__(), (0, 1))
self.assertRaises(ZeroDivisionError, i.__next__)
@support.requires_resource('walltime')
def test_map_timeout(self):
results = []
try:

3
Lib/test/test_concurrent_futures/test_wait.py
View File

@ -3,6 +3,7 @@ import threading
import time
import unittest
from concurrent import futures
from test import support
from .util import (
CANCELLED_FUTURE, CANCELLED_AND_NOTIFIED_FUTURE, EXCEPTION_FUTURE,
@ -53,6 +54,7 @@ class WaitTests:
finished)
self.assertEqual(set([future1]), pending)
@support.requires_resource('walltime')
def test_first_exception(self):
future1 = self.executor.submit(mul, 2, 21)
future2 = self.executor.submit(sleep_and_raise, 1.5)
@ -110,6 +112,7 @@ class WaitTests:
future2]), finished)
self.assertEqual(set(), pending)
@support.requires_resource('walltime')
def test_timeout(self):
future1 = self.executor.submit(mul, 6, 7)
future2 = self.executor.submit(time.sleep, 6)

1
Lib/test/test_eintr.py
View File

@ -9,6 +9,7 @@ from test.support import script_helper
class EINTRTests(unittest.TestCase):
@unittest.skipUnless(hasattr(signal, "setitimer"), "requires setitimer()")
@support.requires_resource('walltime')
def test_all(self):
# Run the tester in a sub-process, to make sure there is only one
# thread (for reliable signal delivery).

1
Lib/test/test_faulthandler.py
View File

@ -683,6 +683,7 @@ class FaultHandlerTests(unittest.TestCase):
with tempfile.TemporaryFile('wb+') as fp:
self.check_dump_traceback_later(fd=fp.fileno())
@support.requires_resource('walltime')
def test_dump_traceback_later_twice(self):
self.check_dump_traceback_later(loops=2)

1
Lib/test/test_httplib.py
View File

@ -1954,6 +1954,7 @@ class HTTPSTest(TestCase):
h.close()
self.assertIn('nginx', server_string)
@support.requires_resource('walltime')
def test_networked_bad_cert(self):
# We feed a "CA" cert that is unrelated to the server's cert
import ssl

5
Lib/test/test_imaplib.py
View File

@ -10,7 +10,7 @@ import calendar
import threading
import socket
from test.support import verbose, run_with_tz, run_with_locale, cpython_only
from test.support import verbose, run_with_tz, run_with_locale, cpython_only, requires_resource
from test.support import hashlib_helper
from test.support import threading_helper
import unittest
@ -456,6 +456,7 @@ class NewIMAPTestsMixin():
with self.imap_class(*server.server_address):
pass
@requires_resource('walltime')
def test_imaplib_timeout_test(self):
_, server = self._setup(SimpleIMAPHandler)
addr = server.server_address[1]
@ -549,6 +550,7 @@ class NewIMAPSSLTests(NewIMAPTestsMixin, unittest.TestCase):
imap_class = IMAP4_SSL
server_class = SecureTCPServer
@requires_resource('walltime')
def test_ssl_raises(self):
ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
self.assertEqual(ssl_context.verify_mode, ssl.CERT_REQUIRED)
@ -563,6 +565,7 @@ class NewIMAPSSLTests(NewIMAPTestsMixin, unittest.TestCase):
ssl_context=ssl_context)
client.shutdown()
@requires_resource('walltime')
def test_ssl_verified(self):
ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
ssl_context.load_verify_locations(CAFILE)

6
Lib/test/test_io.py
View File

@ -4468,10 +4468,12 @@ class CMiscIOTest(MiscIOTest):
self.assertFalse(err.strip('.!'))
@threading_helper.requires_working_threading()
@support.requires_resource('walltime')
def test_daemon_threads_shutdown_stdout_deadlock(self):
self.check_daemon_threads_shutdown_deadlock('stdout')
@threading_helper.requires_working_threading()
@support.requires_resource('walltime')
def test_daemon_threads_shutdown_stderr_deadlock(self):
self.check_daemon_threads_shutdown_deadlock('stderr')
@ -4645,11 +4647,13 @@ class SignalsTest(unittest.TestCase):
os.close(r)
@requires_alarm
@support.requires_resource('walltime')
def test_interrupted_read_retry_buffered(self):
self.check_interrupted_read_retry(lambda x: x.decode('latin1'),
mode="rb")
@requires_alarm
@support.requires_resource('walltime')
def test_interrupted_read_retry_text(self):
self.check_interrupted_read_retry(lambda x: x,
mode="r", encoding="latin1")
@ -4723,10 +4727,12 @@ class SignalsTest(unittest.TestCase):
raise
@requires_alarm
@support.requires_resource('walltime')
def test_interrupted_write_retry_buffered(self):
self.check_interrupted_write_retry(b"x", mode="wb")
@requires_alarm
@support.requires_resource('walltime')
def test_interrupted_write_retry_text(self):
self.check_interrupted_write_retry("x", mode="w", encoding="latin1")

1
Lib/test/test_logging.py
View File

@ -680,6 +680,7 @@ class HandlerTest(BaseTest):
support.is_emscripten, "Emscripten cannot fstat unlinked files."
)
@threading_helper.requires_working_threading()
@support.requires_resource('walltime')
def test_race(self):
# Issue #14632 refers.
def remove_loop(fname, tries):

3
Lib/test/test_poll.py
View File

@ -8,7 +8,7 @@ import threading
import time
import unittest
from test.support import (
cpython_only, requires_subprocess, requires_working_socket
cpython_only, requires_subprocess, requires_working_socket, requires_resource
)
from test.support import threading_helper
from test.support.os_helper import TESTFN
@ -124,6 +124,7 @@ class PollTests(unittest.TestCase):
# select(), modified to use poll() instead.
@requires_subprocess()
@requires_resource('walltime')
def test_poll2(self):
cmd = 'for i in 0 1 2 3 4 5 6 7 8 9; do echo testing...; sleep 1; done'
proc = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE,

1
Lib/test/test_signal.py
View File

@ -745,6 +745,7 @@ class SiginterruptTest(unittest.TestCase):
interrupted = self.readpipe_interrupted(True)
self.assertTrue(interrupted)
@support.requires_resource('walltime')
def test_siginterrupt_off(self):
# If a signal handler is installed and siginterrupt is called with
# a false value for the second argument, when that signal arrives, it

1
Lib/test/test_smtpnet.py
View File

@ -61,6 +61,7 @@ class SmtpSSLTest(unittest.TestCase):
server.ehlo()
server.quit()
@support.requires_resource('walltime')
def test_connect_using_sslcontext(self):
context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
context.check_hostname = False

2
Lib/test/test_ssl.py
View File

@ -2182,6 +2182,7 @@ class NetworkedTests(unittest.TestCase):
self.assertIn(rc, (errno.EAGAIN, errno.EWOULDBLOCK))
@unittest.skipUnless(socket_helper.IPV6_ENABLED, 'Needs IPv6')
@support.requires_resource('walltime')
def test_get_server_certificate_ipv6(self):
with socket_helper.transient_internet('ipv6.google.com'):
_test_get_server_certificate(self, 'ipv6.google.com', 443)
@ -2740,6 +2741,7 @@ def try_protocol_combo(server_protocol, client_protocol, expect_success,
class ThreadedTests(unittest.TestCase):
@support.requires_resource('walltime')
def test_echo(self):
"""Basic test of an SSL client connecting to a server"""
if support.verbose:

2
Lib/test/test_subprocess.py
View File

@ -269,6 +269,7 @@ class ProcessTestCase(BaseTestCase):
self.assertIn('stdin', c.exception.args[0])
self.assertIn('input', c.exception.args[0])
@support.requires_resource('walltime')
def test_check_output_timeout(self):
# check_output() function with timeout arg
with self.assertRaises(subprocess.TimeoutExpired) as c:
@ -1643,6 +1644,7 @@ class RunFuncTestCase(BaseTestCase):
self.assertIn('stdin', c.exception.args[0])
self.assertIn('input', c.exception.args[0])
@support.requires_resource('walltime')
def test_check_output_timeout(self):
with self.assertRaises(subprocess.TimeoutExpired) as c:
cp = self.run_python((

5
Lib/test/test_urllib2net.py
View File

@ -133,6 +133,7 @@ class OtherNetworkTests(unittest.TestCase):
# XXX The rest of these tests aren't very good -- they don't check much.
# They do sometimes catch some major disasters, though.
@support.requires_resource('walltime')
def test_ftp(self):
# Testing the same URL twice exercises the caching in CacheFTPHandler
urls = [
@ -196,6 +197,7 @@ class OtherNetworkTests(unittest.TestCase):
self.assertEqual(res.geturl(),
"http://www.pythontest.net/index.html#frag")
@support.requires_resource('walltime')
def test_redirect_url_withfrag(self):
redirect_url_with_frag = "http://www.pythontest.net/redir/with_frag/"
with socket_helper.transient_internet(redirect_url_with_frag):
@ -334,6 +336,7 @@ class TimeoutTest(unittest.TestCase):
FTP_HOST = 'ftp://www.pythontest.net/'
@support.requires_resource('walltime')
def test_ftp_basic(self):
self.assertIsNone(socket.getdefaulttimeout())
with socket_helper.transient_internet(self.FTP_HOST, timeout=None):
@ -352,6 +355,7 @@ class TimeoutTest(unittest.TestCase):
socket.setdefaulttimeout(None)
self.assertEqual(u.fp.fp.raw._sock.gettimeout(), 60)
@support.requires_resource('walltime')
def test_ftp_no_timeout(self):
self.assertIsNone(socket.getdefaulttimeout())
with socket_helper.transient_internet(self.FTP_HOST):
@ -363,6 +367,7 @@ class TimeoutTest(unittest.TestCase):
socket.setdefaulttimeout(None)
self.assertIsNone(u.fp.fp.raw._sock.gettimeout())
@support.requires_resource('walltime')
def test_ftp_timeout(self):
with socket_helper.transient_internet(self.FTP_HOST):
u = _urlopen_with_retry(self.FTP_HOST, timeout=60)

2
Lib/test/test_urllibnet.py
View File

@ -109,6 +109,7 @@ class urlopenNetworkTests(unittest.TestCase):
open_url.close()
self.assertEqual(code, 404)
@support.requires_resource('walltime')
def test_bad_address(self):
# Make sure proper exception is raised when connecting to a bogus
# address.
@ -191,6 +192,7 @@ class urlretrieveNetworkTests(unittest.TestCase):
logo = "http://www.pythontest.net/"
@support.requires_resource('walltime')
def test_data_header(self):
with self.urlretrieve(self.logo) as (file_location, fileheaders):
datevalue = fileheaders.get('Date')

9
Lib/test/test_xmlrpc.py
View File

@ -1037,38 +1037,47 @@ class MultiPathServerTestCase(BaseServerTestCase):
self.assertEqual(p.add(6,8), 6+8)
self.assertRaises(xmlrpclib.Fault, p.pow, 6, 8)
@support.requires_resource('walltime')
def test_path3(self):
p = xmlrpclib.ServerProxy(URL+"/is/broken")
self.assertRaises(xmlrpclib.Fault, p.add, 6, 8)
@support.requires_resource('walltime')
def test_invalid_path(self):
p = xmlrpclib.ServerProxy(URL+"/invalid")
self.assertRaises(xmlrpclib.Fault, p.add, 6, 8)
@support.requires_resource('walltime')
def test_path_query_fragment(self):
p = xmlrpclib.ServerProxy(URL+"/foo?k=v#frag")
self.assertEqual(p.test(), "/foo?k=v#frag")
@support.requires_resource('walltime')
def test_path_fragment(self):
p = xmlrpclib.ServerProxy(URL+"/foo#frag")
self.assertEqual(p.test(), "/foo#frag")
@support.requires_resource('walltime')
def test_path_query(self):
p = xmlrpclib.ServerProxy(URL+"/foo?k=v")
self.assertEqual(p.test(), "/foo?k=v")
@support.requires_resource('walltime')
def test_empty_path(self):
p = xmlrpclib.ServerProxy(URL)
self.assertEqual(p.test(), "/RPC2")
@support.requires_resource('walltime')
def test_root_path(self):
p = xmlrpclib.ServerProxy(URL + "/")
self.assertEqual(p.test(), "/")
@support.requires_resource('walltime')
def test_empty_path_query(self):
p = xmlrpclib.ServerProxy(URL + "?k=v")
self.assertEqual(p.test(), "?k=v")
@support.requires_resource('walltime')
def test_empty_path_fragment(self):
p = xmlrpclib.ServerProxy(URL + "#frag")
self.assertEqual(p.test(), "#frag")