Update Lib/test/test_support.py from CPython v3.12.0a0

jopemachine · jopemachine · commit 4a76a56fd37b · 2022-09-13T18:35:25.000+09:00
diff --git a/Lib/test/test_support.py b/Lib/test/test_support.py
@@ -9,7 +9,6 @@
 import sys
 import tempfile
 import textwrap
-import time
 import unittest
 import warnings
 
@@ -123,15 +122,18 @@ def test_forget(self):
             os_helper.unlink(mod_filename)
             os_helper.rmtree('__pycache__')
 
+    @support.requires_working_socket()
     def test_HOST(self):
         s = socket.create_server((socket_helper.HOST, 0))
         s.close()
 
+    @support.requires_working_socket()
     def test_find_unused_port(self):
         port = socket_helper.find_unused_port()
         s = socket.create_server((socket_helper.HOST, port))
         s.close()
 
+    @support.requires_working_socket()
     def test_bind_port(self):
         s = socket.socket()
         socket_helper.bind_port(s)
@@ -198,7 +200,7 @@ def test_temp_dir__existing_dir__quiet_true(self):
                                         f'temporary directory {path!r}: '),
                         warn)
 
-    @unittest.skipUnless(hasattr(os, "fork"), "test requires os.fork")
+    @support.requires_fork()
     def test_temp_dir__forked_child(self):
         """Test that a forked child process does not remove the directory."""
         # See bpo-30028 for details.
@@ -308,7 +310,6 @@ def test_temp_cwd__name_none(self):
     def test_sortdict(self):
         self.assertEqual(support.sortdict({3:3, 2:2, 1:1}), "{1: 1, 2: 2, 3: 3}")
 
-    @unittest.skipIf(sys.platform.startswith("win"), "TODO: RUSTPYTHON; actual c fds on windows")
     def test_make_bad_fd(self):
         fd = os_helper.make_bad_fd()
         with self.assertRaises(OSError) as cm:
@@ -429,9 +430,14 @@ def test_check__all__(self):
                              extra=extra,
                              not_exported=not_exported)
 
-        extra = {'TextTestResult', 'installHandler'}
+        extra = {
+            'TextTestResult',
+            'findTestCases',
+            'getTestCaseNames',
+            'installHandler',
+            'makeSuite',
+        }
         not_exported = {'load_tests', "TestProgram", "BaseTestSuite"}
-
         support.check__all__(self,
                              unittest,
                              ("unittest.result", "unittest.case",
@@ -447,6 +453,7 @@ def test_check__all__(self):
     @unittest.expectedFailure
     @unittest.skipUnless(hasattr(os, 'waitpid') and hasattr(os, 'WNOHANG'),
                          'need os.waitpid() and os.WNOHANG')
+    @support.requires_fork()
     def test_reap_children(self):
         # Make sure that there is no other pending child process
         support.reap_children()
@@ -457,33 +464,20 @@ def test_reap_children(self):
             # child process: do nothing, just exit
             os._exit(0)
 
-        t0 = time.monotonic()
-        deadline = time.monotonic() + support.SHORT_TIMEOUT
-
         was_altered = support.environment_altered
         try:
             support.environment_altered = False
             stderr = io.StringIO()
 
-            while True:
-                if time.monotonic() > deadline:
-                    self.fail("timeout")
-
-                old_stderr = sys.__stderr__
-                try:
-                    sys.__stderr__ = stderr
+            for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
+                with support.swap_attr(support.print_warning, 'orig_stderr', stderr):
                     support.reap_children()
-                finally:
-                    sys.__stderr__ = old_stderr
 
                 # Use environment_altered to check if reap_children() found
                 # the child process
                 if support.environment_altered:
                     break
 
-                # loop until the child process completed
-                time.sleep(0.100)
-
             msg = "Warning -- reap_children() reaped child process %s" % pid
             self.assertIn(msg, stderr.getvalue())
             self.assertTrue(support.environment_altered)
@@ -494,6 +488,7 @@ def test_reap_children(self):
         # pending child process
         support.reap_children()
 
+    @support.requires_subprocess()
     def check_options(self, args, func, expected=None):
         code = f'from test.support import {func}; print(repr({func}()))'
         cmd = [sys.executable, *args, '-c', code]
@@ -509,6 +504,8 @@ def check_options(self, args, func, expected=None):
         self.assertEqual(proc.stdout.rstrip(), repr(expected))
         self.assertEqual(proc.returncode, 0)
 
+    # TODO: RUSTPYTHON
+    @unittest.expectedFailure
     def test_args_from_interpreter_flags(self):
         # Test test.support.args_from_interpreter_flags()
         for opts in (
@@ -521,6 +518,7 @@ def test_args_from_interpreter_flags(self):
             ['-E'],
             ['-v'],
             ['-b'],
+            ['-P'],
             ['-q'],
             ['-I'],
             # same option multiple times
@@ -540,7 +538,8 @@ def test_args_from_interpreter_flags(self):
             with self.subTest(opts=opts):
                 self.check_options(opts, 'args_from_interpreter_flags')
 
-        self.check_options(['-I', '-E', '-s'], 'args_from_interpreter_flags',
+        self.check_options(['-I', '-E', '-s', '-P'],
+                           'args_from_interpreter_flags',
                            ['-I'])
 
     def test_optim_args_from_interpreter_flags(self):
@@ -660,11 +659,16 @@ def id(self):
             self.assertFalse(support.match_test(test_access))
             self.assertTrue(support.match_test(test_chdir))
 
+
     @unittest.skipIf(sys.platform.startswith("win"), "TODO: RUSTPYTHON; os.dup on windows")
+    @unittest.skipIf(support.is_emscripten, "Unstable in Emscripten")
+    @unittest.skipIf(support.is_wasi, "Unavailable on WASI")
     def test_fd_count(self):
         # We cannot test the absolute value of fd_count(): on old Linux
         # kernel or glibc versions, os.urandom() keeps a FD open on
         # /dev/urandom device and Python has 4 FD opens instead of 3.
+        # Test is unstable on Emscripten. The platform starts and stops
+        # background threads that use pipes and epoll fds.
         start = os_helper.fd_count()
         fd = os.open(__file__, os.O_RDONLY)
         try:
@@ -675,14 +679,8 @@ def test_fd_count(self):
 
     def check_print_warning(self, msg, expected):
         stderr = io.StringIO()
-
-        old_stderr = sys.__stderr__
-        try:
-            sys.__stderr__ = stderr
+        with support.swap_attr(support.print_warning, 'orig_stderr', stderr):
             support.print_warning(msg)
-        finally:
-            sys.__stderr__ = old_stderr
-
         self.assertEqual(stderr.getvalue(), expected)
 
     def test_print_warning(self):
@@ -691,6 +689,14 @@ def test_print_warning(self):
         self.check_print_warning("a\nb",
                                  'Warning -- a\nWarning -- b\n')
 
+    # TODO: RUSTPYTHON
+    @unittest.expectedFailure
+    def test_has_strftime_extensions(self):
+        if support.is_emscripten or sys.platform == "win32":
+            self.assertFalse(support.has_strftime_extensions)
+        else:
+            self.assertTrue(support.has_strftime_extensions)
+
     # XXX -follows a list of untested API
     # make_legacy_pyc
     # is_resource_enabled
@@ -716,4 +722,4 @@ def test_print_warning(self):
 
 
 if __name__ == '__main__':
-    unittest.main()
+    unittest.main()
