[pull] main from python:main

Lib/test/test_capi/test_opt.py

@@ -5127,6 +5127,16 @@ def f():
                               f" {executor} at offset {idx} rather"
                               f" than expected _EXIT_TRACE")
 
+    def test_jit_shutdown_after_cold_executor_creation(self):
+        script_helper.assert_python_ok("-c", textwrap.dedent(f"""
+            def f():
+                for x in range({TIER2_THRESHOLD + 3}):
+                    for y in range({TIER2_THRESHOLD + 3}):
+                        z = x + y
+
+            f()
+        """), PYTHON_JIT="1")
+
     def test_enter_executor_valid_op_arg(self):
         script_helper.assert_python_ok("-c", textwrap.dedent("""
             import sys

Lib/test/test_concurrent_futures/util.py

@@ -139,6 +139,8 @@ def get_context(self):
             self.skipTest("require unix system")
         if support.check_sanitizer(thread=True):
             self.skipTest("TSAN doesn't support threads after fork")
+        if "forkserver" not in multiprocessing.get_all_start_methods():
+            self.skipTest("forkserver start method is not available")
         return super().get_context()
 
     def create_event(self):

Lib/test/test_logging.py

@@ -6704,7 +6704,14 @@ def add_record(message: str) -> None:
         self.assertTrue(found, msg=msg)
 
     def test_rollover_at_midnight(self, weekly=False):
+        # Create the log file in a fresh directory under a never used name:
+        # on Windows, NTFS file tunneling restores the original creation time
+        # of a file recreated with the same name.
         os_helper.unlink(self.fn)
+        dirname = tempfile.mkdtemp()
+        self.addCleanup(os_helper.rmtree, dirname)
+        self.fn = os.path.join(dirname, 'test_rollover.log')
+
         # Emit the first records a little after the beginning of a whole
         # second, so that their file times fall inside that second and not the
         # previous one, which would cause an unwanted rollover.
@@ -6730,7 +6737,24 @@ def test_rollover_at_midnight(self, weekly=False):
         # changed, so the rollover cannot be forced by back-dating the file.
         # Wait until the clock reaches a rollover time set one second ahead.
         rollover = int(time.time()) + 1
+        if rollover - time.time() < 0.1:
+            # Leave time to emit a record before the rollover time.
+            rollover += 1
         atTime = datetime.datetime.fromtimestamp(rollover).time()
+        rolloverDate = (datetime.datetime.fromtimestamp(rollover)
+                        - datetime.timedelta(days=7 if weekly else 1))
+        otherfn = f'{self.fn}.{rolloverDate:%Y-%m-%d}'
+
+        # A record emitted before the rollover time is not rolled over.
+        fh = logging.handlers.TimedRotatingFileHandler(
+            self.fn, encoding="utf-8", when=when, atTime=atTime)
+        fh.setFormatter(fmt)
+        r2 = logging.makeLogRecord({'msg': 'testing1 3'})
+        fh.emit(r2)
+        fh.close()
+        self.assertFalse(os.path.exists(otherfn),
+                         msg=f'{otherfn} was rolled over too early')
+
         while time.time() < rollover:
             time.sleep(rollover - time.time())
         for i in range(2):
@@ -6740,9 +6764,6 @@ def test_rollover_at_midnight(self, weekly=False):
             r2 = logging.makeLogRecord({'msg': f'testing2 {i}'})
             fh.emit(r2)
             fh.close()
-        rolloverDate = (datetime.datetime.fromtimestamp(rollover)
-                        - datetime.timedelta(days=7 if weekly else 1))
-        otherfn = f'{self.fn}.{rolloverDate:%Y-%m-%d}'
         self.assertLogFile(otherfn)
         with open(self.fn, encoding="utf-8") as f:
             for i, line in enumerate(f):

Lib/test/test_multiprocessing_forkserver/__init__.py

@@ -1,3 +1,4 @@
+import multiprocessing
 import os.path
 import sys
 import unittest
@@ -12,5 +13,10 @@
 if not support.has_fork_support:
     raise unittest.SkipTest("requires working os.fork()")
 
+# The forkserver start method requires passing file descriptors over a Unix
+# socket, which is not available on every platform (e.g. Solaris/illumos).
+if "forkserver" not in multiprocessing.get_all_start_methods():
+    raise unittest.SkipTest("forkserver start method is not available")
+
 def load_tests(*args):
     return support.load_package_tests(os.path.dirname(__file__), *args)

Lib/test/test_os/test_os.py

@@ -156,7 +156,8 @@ def test_getcwd_long_path(self):
                         # ("The filename or extension is too long")
                         break
                     except OSError as exc:
-                        if exc.errno == errno.ENAMETOOLONG:
+                        # DragonFly BSD raises EFAULT for a too long path.
+                        if exc.errno in (errno.ENAMETOOLONG, errno.EFAULT):
                             break
                         else:
                             raise
@@ -1058,11 +1059,18 @@ def support_subsecond(self, filename):
                 or (st.st_mtime != st[8])
                 or (st.st_ctime != st[9]))
 
+    def support_atime(self, filename):
+        # Heuristic to check if the filesystem stores the access time.
+        # Use whole seconds, to not depend on the timestamp resolution.
+        os.utime(filename, (1, 2))
+        return os.stat(filename).st_atime == 1
+
     def _test_utime(self, set_time, filename=None):
         if not filename:
             filename = self.fname
 
         support_subsecond = self.support_subsecond(filename)
+        support_atime = self.support_atime(filename)
         if support_subsecond:
             # Timestamp with a resolution of 1 microsecond (10^-6).
             #
@@ -1089,18 +1097,22 @@ def _test_utime(self, set_time, filename=None):
             # digits worth of sub-second precision.
             # Some day it would be good to fix this upstream.
             delta=1e-5
-            self.assertAlmostEqual(st.st_atime, atime_ns * 1e-9, delta=1e-5)
+            if support_atime:
+                self.assertAlmostEqual(st.st_atime, atime_ns * 1e-9, delta=1e-5)
+                self.assertAlmostEqual(st.st_atime_ns, atime_ns, delta=1e9 * 1e-5)
             self.assertAlmostEqual(st.st_mtime, mtime_ns * 1e-9, delta=1e-5)
-            self.assertAlmostEqual(st.st_atime_ns, atime_ns, delta=1e9 * 1e-5)
             self.assertAlmostEqual(st.st_mtime_ns, mtime_ns, delta=1e9 * 1e-5)
         else:
             if support_subsecond:
-                self.assertAlmostEqual(st.st_atime, atime_ns * 1e-9, delta=1e-6)
+                if support_atime:
+                    self.assertAlmostEqual(st.st_atime, atime_ns * 1e-9, delta=1e-6)
                 self.assertAlmostEqual(st.st_mtime, mtime_ns * 1e-9, delta=1e-6)
             else:
-                self.assertEqual(st.st_atime, atime_ns * 1e-9)
+                if support_atime:
+                    self.assertEqual(st.st_atime, atime_ns * 1e-9)
                 self.assertEqual(st.st_mtime, mtime_ns * 1e-9)
-            self.assertEqual(st.st_atime_ns, atime_ns)
+            if support_atime:
+                self.assertEqual(st.st_atime_ns, atime_ns)
             self.assertEqual(st.st_mtime_ns, mtime_ns)
 
     def test_utime(self):

Lib/test/test_os/test_posix.py

@@ -46,7 +46,8 @@ def _supports_sched():
     try:
         posix.sched_getscheduler(0)
     except OSError as e:
-        if e.errno == errno.ENOSYS:
+        # DragonFly BSD requires privileges to use the scheduler API.
+        if e.errno in (errno.ENOSYS, errno.EPERM):
             return False
     return True
 
@@ -823,8 +824,7 @@ def test_makedev(self):
         # a special case for NODEV, on others this is just an implementation
         # artifact.
         if (hasattr(posix, 'NODEV') and
-            sys.platform.startswith(('linux', 'macos', 'freebsd', 'dragonfly',
-                                     'sunos'))):
+            sys.platform.startswith(('linux', 'macos', 'freebsd', 'sunos'))):
             NODEV = posix.NODEV
             self.assertEqual(posix.major(NODEV), NODEV)
             self.assertEqual(posix.minor(NODEV), NODEV)
@@ -1453,6 +1453,7 @@ def test_bug_140634(self):
         del sched_priority, param  # should not crash
         support.gc_collect()  # just to be sure
 
+    @requires_sched
     @unittest.skipUnless(hasattr(posix, "sched_rr_get_interval"), "no function")
     def test_sched_rr_get_interval(self):
         try:

Lib/test/test_shlex.py

@@ -1,8 +1,11 @@
 import io
 import itertools
+import os
 import shlex
 import string
+import tempfile
 import unittest
+from unittest.mock import patch
 from test.support import cpython_only
 from test.support import import_helper
 
@@ -376,6 +379,203 @@ def testPunctuationCharsReadOnly(self):
         with self.assertRaises(AttributeError):
             shlex_instance.punctuation_chars = False
 
+    def testLinenoAfterNewLine(self):
+        s = shlex.shlex("line 1\nline 2")
+        self.assertEqual(s.lineno, 1)  # before consumption
+        list(s)
+        self.assertEqual(s.lineno, 2)
+
+    def testLinenoAfterComment(self):
+        """Comment handler increments lineno even without a trailing newline."""
+        s = shlex.shlex("line 1 # line 2")
+        list(s)
+        self.assertEqual(s.lineno, 2)
+
+    def testPushToken(self):
+        s = shlex.shlex("b c")
+        s.push_token("a")
+        self.assertListEqual(list(s), ["a", "b", "c"])
+
+    def testPushTokenLifo(self):
+        s = shlex.shlex("")
+        s.push_token("first")
+        s.push_token("last")
+        self.assertListEqual(list(s), ["last", "first"])
+
+    def testPushTokenDebug(self):
+        s = shlex.shlex("")
+        s.debug = 1
+        tok = "a"
+        with patch("builtins.print") as mock_print:
+            s.push_token(tok)
+        mock_print.assert_called_once_with(f"shlex: pushing token {tok!r}")
+
+    def testPushSourceString(self):
+        s = shlex.shlex("world")
+        s.push_source("hello")
+        self.assertListEqual(list(s), ["hello", "world"])
+
+    def testPushSourceStream(self):
+        s = shlex.shlex("world")
+        s.push_source(io.StringIO("hello"))
+        self.assertListEqual(list(s), ["hello", "world"])
+
+    def testPushSourceStreamDebug(self):
+        s = shlex.shlex("")
+        stream = io.StringIO("hello")
+        s.debug = 1
+        with patch("builtins.print") as mock_print:
+            s.push_source(stream)
+        mock_print.assert_called_once_with(f"shlex: pushing to stream {stream}")
+
+    def testPushSourceNewfile(self):
+        """shlex.push_source sets infile to newfile; pop_source restores the original on exhaustion."""
+        original_file = "original.sh"
+        new_file = "new.sh"
+        s = shlex.shlex("b", infile=original_file)
+        s.debug = 1
+        with patch("builtins.print") as mock_print:
+            s.push_source("a", newfile=new_file)
+        mock_print.assert_called_once_with(f"shlex: pushing to file {new_file}")
+        self.assertEqual(s.infile, new_file)
+        s.debug = 0
+        list(s)
+        self.assertEqual(s.infile, original_file)
+
+    def testPopSourceDebug(self):
+        """pop_source emits debug output when debug is set."""
+        s = shlex.shlex("b")
+        original_stream = s.instream
+        s.push_source("a")
+        s.debug = 1
+        with patch("builtins.print") as mock_print:
+            list(s)  # exhausts pushed source and triggers pop_source internally
+        mock_print.assert_any_call(f"shlex: popping to {original_stream}, line 1")
+
+    def testErrorLeaderTracksPosition(self):
+        infile_label = "test.sh"
+        s = shlex.shlex("line 1\nline 2", infile=infile_label)
+        list(s)
+        result = s.error_leader()
+        self.assertEqual(result, f'"{infile_label}", line 2: ')
+
+    def testErrorLeaderOverrides(self):
+        s = shlex.shlex("foo", infile="original.sh")
+        infile_label_override = "override.sh"
+        lineno_override = 42
+        result = s.error_leader(infile=infile_label_override, lineno=lineno_override)
+        self.assertEqual(result, f'"{infile_label_override}", line {lineno_override}: ')
+
+    def testNoClosingQuotation(self):
+        s = shlex.shlex('"foo')
+        with self.assertRaisesRegex(ValueError, "No closing quotation"):
+            list(s)
+
+    def testNoEscapedCharacter(self):
+        s = shlex.shlex("\\", posix=True)
+        with self.assertRaisesRegex(ValueError, "No escaped character"):
+            list(s)
+
+    def testSourcehookStripsQuotes(self):
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".sh", delete_on_close=False) as f:
+            f.write("hello")
+            f.close()
+            s = shlex.shlex("")
+            newfile, stream = s.sourcehook(f'"{f.name}"')
+            stream.close()
+        self.assertEqual(newfile, f.name)
+
+    def testSourcehookAbsolutePath(self):
+        with tempfile.NamedTemporaryFile(mode="w", delete_on_close=False) as f:
+            f.close()
+            s = shlex.shlex("", infile="/some/dir/main.sh")
+            newfile, stream = s.sourcehook(f.name)
+            stream.close()
+        self.assertEqual(newfile, f.name)
+
+    def testSourcehookRelativePath(self):
+        with tempfile.TemporaryDirectory() as d:
+            fpath = os.path.join(d, "included.sh")
+            with open(fpath, "w"):
+                pass
+            s = shlex.shlex("", infile=os.path.join(d, "main.sh"))
+            newfile, stream = s.sourcehook("included.sh")
+            stream.close()
+            self.assertEqual(newfile, fpath)
+
+    def testSourceInclusion(self):
+        """shlex.source sets a trigger keyword: when the lexer reads a token equal
+        to it, the next token is consumed as a filename and passed to
+        sourcehook, which returns a stream to push onto the input stack.
+        Tokens flow from that stream first, then resume from the original.
+        """
+        s = shlex.shlex("trigger filename remaining")
+        s.source = "trigger"
+        s.sourcehook = lambda f: (f, io.StringIO("included"))
+        self.assertEqual(list(s), ["included", "remaining"])
+
+    def testGetTokenPopsPushbackDebug(self):
+        s = shlex.shlex("")
+        s.push_token("hello")
+        s.debug = 1  # set after push_token to isolate the pop-token branch
+        with patch("builtins.print") as mock_print:
+            tok = s.get_token()
+        self.assertEqual(tok, "hello")
+        mock_print.assert_called_once_with("shlex: popping token 'hello'")
+
+    def testDebugWhitespaceInWhitespaceState(self):
+        s = shlex.shlex(" a")
+        s.debug = 2
+        with patch("builtins.print") as mock_print:
+            list(s)
+        mock_print.assert_any_call("shlex: I see whitespace in whitespace state")
+
+    def testDebugWhitespaceInWordState(self):
+        s = shlex.shlex("a b")
+        s.debug = 2
+        with patch("builtins.print") as mock_print:
+            list(s)
+        mock_print.assert_any_call("shlex: I see whitespace in word state")
+
+    def testDebugPunctuationInWordState(self):
+        s = shlex.shlex("a(")
+        s.debug = 2
+        with patch("builtins.print") as mock_print:
+            list(s)
+        mock_print.assert_any_call("shlex: I see punctuation in word state")
+
+    def testDebugRawToken(self):
+        s = shlex.shlex("hello")
+        s.debug = 2
+        with patch("builtins.print") as mock_print:
+            list(s)
+        mock_print.assert_any_call("shlex: raw token='hello'")
+
+    def testDebugEOFInQuote(self):
+        s = shlex.shlex('"oops', posix=True)
+        s.debug = 2
+        with patch('builtins.print') as mock_print:
+            with self.assertRaises(ValueError):
+                list(s)
+        msgs = [call.args[0] for call in mock_print.call_args_list]
+        self.assertTrue(any("EOF in quotes" in m for m in msgs))
+
+    def testDebugEOFInEscape(self):
+        s = shlex.shlex("oops\\", posix=True)
+        s.debug = 2
+        with patch("builtins.print") as mock_print:
+            with self.assertRaises(ValueError):
+                list(s)
+        msgs = [call.args[0] for call in mock_print.call_args_list]
+        self.assertTrue(any("EOF in escape" in m for m in msgs))
+
+    def testDebugStateTrace(self):
+        s = shlex.shlex("a")
+        s.debug = 3
+        with patch("builtins.print") as mock_print:
+            list(s)
+        mock_print.assert_any_call("shlex: in state ' ' I see character: 'a'")
+
     @cpython_only
     def test_lazy_imports(self):
         import_helper.ensure_lazy_imports('shlex', {'collections', 're', 'os'})