mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-09-01 14:38:00 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

gh-109162: libregrtest: fix _decode_worker_job() (#109202)

Browse source 
Decode also HuntRefleak() object inside the RunTests object.

Add an unit test on huntrleaks with multiprocessing (-R -jN).
This commit is contained in:
Victor Stinner 2023-09-10 00:51:24 +02:00 • committed by GitHub
parent d3ed9921cd
commit 24fa8f2046
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 21 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

6
Lib/test/libregrtest/runtest.py
View file

@ -249,6 +249,12 @@ class RunTests:
else: else:
yield from self.tests yield from self.tests
@staticmethod
def from_json_dict(json_dict):
if json_dict['hunt_refleak']:
json_dict['hunt_refleak'] = HuntRefleak(**json_dict['hunt_refleak'])
return RunTests(**json_dict)
# Minimum duration of a test to display its duration or to mention that # Minimum duration of a test to display its duration or to mention that
# the test is running in background # the test is running in background

2
Lib/test/libregrtest/runtest_mp.py
View file

@ -68,7 +68,7 @@ class _EncodeWorkerJob(json.JSONEncoder):
def _decode_worker_job(d: dict[str, Any]) -> WorkerJob | dict[str, Any]: def _decode_worker_job(d: dict[str, Any]) -> WorkerJob | dict[str, Any]:
if "__worker_job__" in d: if "__worker_job__" in d:
d.pop('__worker_job__') d.pop('__worker_job__')
d['runtests'] = RunTests(**d['runtests']) d['runtests'] = RunTests.from_json_dict(d['runtests'])
return WorkerJob(**d) return WorkerJob(**d)
if "__namespace__" in d: if "__namespace__" in d:
d.pop('__namespace__') d.pop('__namespace__')

18
Lib/test/test_regrtest.py
View file

@ -1018,12 +1018,16 @@ class ArgsTestCase(BaseTestCase):
stats=TestStats(4, 1), stats=TestStats(4, 1),
forever=True) forever=True)
def check_leak(self, code, what): def check_leak(self, code, what, *, multiprocessing=False):
test = self.create_test('huntrleaks', code=code) test = self.create_test('huntrleaks', code=code)
filename = 'reflog.txt' filename = 'reflog.txt'
self.addCleanup(os_helper.unlink, filename) self.addCleanup(os_helper.unlink, filename)
output = self.run_tests('--huntrleaks', '6:3:', test, cmd = ['--huntrleaks', '6:3:']
if multiprocessing:
cmd.append('-j1')
cmd.append(test)
output = self.run_tests(*cmd,
exitcode=EXITCODE_BAD_TEST, exitcode=EXITCODE_BAD_TEST,
stderr=subprocess.STDOUT) stderr=subprocess.STDOUT)
self.check_executed_tests(output, [test], failed=test, stats=1) self.check_executed_tests(output, [test], failed=test, stats=1)
@ -1039,7 +1043,7 @@ class ArgsTestCase(BaseTestCase):
self.assertIn(line2, reflog) self.assertIn(line2, reflog)
@unittest.skipUnless(support.Py_DEBUG, 'need a debug build') @unittest.skipUnless(support.Py_DEBUG, 'need a debug build')
def test_huntrleaks(self): def check_huntrleaks(self, *, multiprocessing: bool):
# test --huntrleaks # test --huntrleaks
code = textwrap.dedent(""" code = textwrap.dedent("""
import unittest import unittest
@ -1050,7 +1054,13 @@ class ArgsTestCase(BaseTestCase):
def test_leak(self): def test_leak(self):
GLOBAL_LIST.append(object()) GLOBAL_LIST.append(object())
""") """)
self.check_leak(code, 'references') self.check_leak(code, 'references', multiprocessing=multiprocessing)
def test_huntrleaks(self):
self.check_huntrleaks(multiprocessing=False)
def test_huntrleaks_mp(self):
self.check_huntrleaks(multiprocessing=True)
@unittest.skipUnless(support.Py_DEBUG, 'need a debug build') @unittest.skipUnless(support.Py_DEBUG, 'need a debug build')
def test_huntrleaks_fd_leak(self): def test_huntrleaks_fd_leak(self):