gh-93353: regrtest checks for leaked temporary files

Lib/test/libregrtest/runtest.py

@@ -80,6 +80,11 @@ class EnvChanged(Failed):
  def __str__(self) -> str:
  return f"{self.name} failed (env changed)"
 
+ # Convert Passed to EnvChanged
+ @staticmethod
+ def from_passed(other):
+ return EnvChanged(other.name, other.duration_sec, other.xml_data)
+
 
 class RefLeak(Failed):
  def __str__(self) -> str:

Lib/test/libregrtest/runtest_mp.py

@@ -1,6 +1,6 @@
 import faulthandler
 import json
-import os
+import os.path
 import queue
 import shlex
 import signal
@@ -17,7 +17,8 @@
 from test.libregrtest.cmdline import Namespace
 from test.libregrtest.main import Regrtest
 from test.libregrtest.runtest import (
- runtest, is_failed, TestResult, Interrupted, Timeout, ChildError, PROGRESS_MIN_TIME)
+ runtest, is_failed, TestResult, Interrupted, Timeout, ChildError,
+ PROGRESS_MIN_TIME, Passed, EnvChanged)
 from test.libregrtest.setup import setup_tests
 from test.libregrtest.utils import format_duration, print_warning
 
@@ -52,7 +53,7 @@ def parse_worker_args(worker_args) -> tuple[Namespace, str]:
  return (ns, test_name)
 
 
-def run_test_in_subprocess(testname: str, ns: Namespace) -> subprocess.Popen:
+def run_test_in_subprocess(testname: str, ns: Namespace, tmp_dir: str) -> subprocess.Popen:
  ns_dict = vars(ns)
  worker_args = (ns_dict, testname)
  worker_args = json.dumps(worker_args)
@@ -66,10 +67,14 @@ def run_test_in_subprocess(testname: str, ns: Namespace) -> subprocess.Popen:
  '-m', 'test.regrtest',
  '--worker-args', worker_args]
 
+ env = dict(os.environ)
+ env['TMPDIR'] = tmp_dir
+ env['TEMPDIR'] = tmp_dir
+
  # Running the child from the same working directory as regrtest's original
  # invocation ensures that TEMPDIR for the child is the same when
  # sysconfig.is_python_build() is true. See issue 15300.
- kw = {}
+ kw = {'env': env}
  if USE_PROCESS_GROUP:
  kw['start_new_session'] = True
  return subprocess.Popen(cmd,
@@ -206,12 +211,12 @@ def mp_result_error(
  test_result.duration_sec = time.monotonic() - self.start_time
  return MultiprocessResult(test_result, stdout, err_msg)
 
- def _run_process(self, test_name: str) -> tuple[int, str, str]:
+ def _run_process(self, test_name: str, tmp_dir: str) -> tuple[int, str, str]:
  self.start_time = time.monotonic()
 
  self.current_test_name = test_name
  try:
- popen = run_test_in_subprocess(test_name, self.ns)
+ popen = run_test_in_subprocess(test_name, self.ns, tmp_dir)
 
  self._killed = False
  self._popen = popen
@@ -266,7 +271,17 @@ def _run_process(self, test_name: str) -> tuple[int, str, str]:
  self.current_test_name = None
 
  def _runtest(self, test_name: str) -> MultiprocessResult:
- retcode, stdout = self._run_process(test_name)
+ # gh-93353: Check for leaked temporary files in the parent process,
+ # since the deletion of temporary files can happen late during
+ # Python finalization: too late for libregrtest.
+ tmp_dir = os.getcwd() + '_tmpdir'
+ tmp_dir = os.path.abspath(tmp_dir)
+ try:
+ os.mkdir(tmp_dir)
+ retcode, stdout = self._run_process(test_name, tmp_dir)
+ finally:
+ tmp_files = os.listdir(tmp_dir)
+ os_helper.rmtree(tmp_dir)
 
  if retcode is None:
  return self.mp_result_error(Timeout(test_name), stdout)
@@ -289,6 +304,14 @@ def _runtest(self, test_name: str) -> MultiprocessResult:
  if err_msg is not None:
  return self.mp_result_error(ChildError(test_name), stdout, err_msg)
 
+ if tmp_files:
+ msg = (f'\n\n'
+ f'Warning -- Test leaked temporary files ({len(tmp_files)}): '
+ f'{", ".join(sorted(tmp_files))}')
+ stdout += msg
+ if isinstance(result, Passed):
+ result = EnvChanged.from_passed(result)
+
  return MultiprocessResult(result, stdout, err_msg)
 
  def run(self) -> None:

Lib/test/test_regrtest.py

@@ -1357,6 +1357,26 @@ def test_cleanup(self):
  for name in names:
  self.assertFalse(os.path.exists(name), name)
 
+ def test_leak_tmp_file(self):
+ code = textwrap.dedent(r"""
+ import os.path
+ import tempfile
+ import unittest
+
+ class FileTests(unittest.TestCase):
+ def test_leak_tmp_file(self):
+ filename = os.path.join(tempfile.gettempdir(), 'mytmpfile')
+ with open(filename, "wb") as fp:
+ fp.write(b'content')
+ """)
+ testname = self.create_test(code=code)
+
+ output = self.run_tests("--fail-env-changed", "-v", "-j1", testname, exitcode=3)
+ self.check_executed_tests(output, [testname],
+ env_changed=[testname],
+ fail_env_changed=True)
+ self.assertIn("Warning -- Test leaked temporary files (1): mytmpfile", output)
+
 
 class TestUtils(unittest.TestCase):
  def test_format_duration(self):