Issue #26325: Added test.support.check_no_resource_warning() to check that

no ResourceWarning is emitted.

4 files changed, 29 insertions, 20 deletions

diff --git a/Lib/test/support/__init__.py b/Lib/test/support/__init__.py
index dfb4c25..b82f9cb 100644
--- a/Lib/test/support/__init__.py
+++ b/Lib/test/support/__init__.py
@@ -100,7 +100,8 @@ __all__ = [
     # threads
     "threading_setup", "threading_cleanup", "reap_threads", "start_threads",
     # miscellaneous
-    "check_warnings", "EnvironmentVarGuard", "run_with_locale", "swap_item",
+    "check_warnings", "check_no_resource_warning", "EnvironmentVarGuard",
+    "run_with_locale", "swap_item",
     "swap_attr", "Matcher", "set_memlimit", "SuppressCrashReport", "sortdict",
     "run_with_tz",
     ]
@@ -1147,6 +1148,27 @@ def check_warnings(*filters, **kwargs):
     return _filterwarnings(filters, quiet)
 
 
+@contextlib.contextmanager
+def check_no_resource_warning(testcase):
+    """Context manager to check that no ResourceWarning is emitted.
+
+    Usage:
+
+        with check_no_resource_warning(self):
+            f = open(...)
+            ...
+            del f
+
+    You must remove the object which may emit ResourceWarning before
+    the end of the context manager.
+    """
+    with warnings.catch_warnings(record=True) as warns:
+        warnings.filterwarnings('always', category=ResourceWarning)
+        yield
+        gc_collect()
+    testcase.assertEqual(warns, [])
+
+
 class CleanImport(object):
     """Context manager to force import to return a new module reference.
 
diff --git a/Lib/test/test_asyncio/test_subprocess.py b/Lib/test/test_asyncio/test_subprocess.py
index e90f17d..c111e2f 100644
--- a/Lib/test/test_asyncio/test_subprocess.py
+++ b/Lib/test/test_asyncio/test_subprocess.py
@@ -427,10 +427,9 @@ class SubprocessMixin:
 
             create = asyncio.create_subprocess_exec(sys.executable, '-c',
                                                     'pass', loop=self.loop)
-            with warnings.catch_warnings(record=True) as warns:
+            with support.check_no_resource_warning(self):
                 with self.assertRaises(exc):
                     self.loop.run_until_complete(create)
-                self.assertEqual(warns, [])
 
 
 if sys.platform != 'win32':
diff --git a/Lib/test/test_io.py b/Lib/test/test_io.py
index 28f440d..86440c5 100644
--- a/Lib/test/test_io.py
+++ b/Lib/test/test_io.py
@@ -681,18 +681,14 @@ class IOTest(unittest.TestCase):
             f2.readline()
 
     def test_nonbuffered_textio(self):
-        with warnings.catch_warnings(record=True) as recorded:
+        with support.check_no_resource_warning(self):
             with self.assertRaises(ValueError):
                 self.open(support.TESTFN, 'w', buffering=0)
-            support.gc_collect()
-        self.assertEqual(recorded, [])
 
     def test_invalid_newline(self):
-        with warnings.catch_warnings(record=True) as recorded:
+        with support.check_no_resource_warning(self):
             with self.assertRaises(ValueError):
                 self.open(support.TESTFN, 'w', newline='invalid')
-            support.gc_collect()
-        self.assertEqual(recorded, [])
 
 
 class CIOTest(IOTest):
@@ -3366,10 +3362,8 @@ class MiscIOTest(unittest.TestCase):
         # When using closefd=False, there's no warning
         r, w = os.pipe()
         fds += r, w
-        with warnings.catch_warnings(record=True) as recorded:
+        with support.check_no_resource_warning(self):
             open(r, *args, closefd=False, **kwargs)
-            support.gc_collect()
-        self.assertEqual(recorded, [])
 
     def test_warn_on_dealloc_fd(self):
         self._check_warn_on_dealloc_fd("rb", buffering=0)
diff --git a/Lib/test/test_xml_etree.py b/Lib/test/test_xml_etree.py
index 358484d..44e3142 100644
--- a/Lib/test/test_xml_etree.py
+++ b/Lib/test/test_xml_etree.py
@@ -567,14 +567,11 @@ class ElementTreeTest(unittest.TestCase):
             self.assertFalse(f.closed)
         self.assertEqual(str(cm.exception), "unknown event 'bogus'")
 
-        with warnings.catch_warnings(record=True) as w:
-            warnings.filterwarnings("always", category=ResourceWarning)
+        with support.check_no_resource_warning(self):
             with self.assertRaises(ValueError) as cm:
                 iterparse(SIMPLE_XMLFILE, events)
             self.assertEqual(str(cm.exception), "unknown event 'bogus'")
             del cm
-            support.gc_collect()
-        self.assertEqual(w, [])
 
         source = io.BytesIO(
             b"<?xml version='1.0' encoding='iso-8859-1'?>\n"
@@ -601,15 +598,12 @@ class ElementTreeTest(unittest.TestCase):
         it = iterparse(TESTFN)
         action, elem = next(it)
         self.assertEqual((action, elem.tag), ('end', 'document'))
-        with warnings.catch_warnings(record=True) as w:
-            warnings.filterwarnings("always", category=ResourceWarning)
+        with support.check_no_resource_warning(self):
             with self.assertRaises(ET.ParseError) as cm:
                 next(it)
             self.assertEqual(str(cm.exception),
                     'junk after document element: line 1, column 12')
             del cm, it
-            support.gc_collect()
-        self.assertEqual(w, [])
 
     def test_writefile(self):
         elem = ET.Element("tag")