Update some libs & tests to 3.14.7 (#8577)

* Update some libs and tests to 3.14.7

* test_shlex.py

* email

Lib/email/utils.py

@@ -317,10 +317,13 @@ def parsedate_to_datetime(data):
     if parsed_date_tz is None:
         raise ValueError('Invalid date value or format "%s"' % str(data))
     *dtuple, tz = parsed_date_tz
-    if tz is None:
-        return datetime.datetime(*dtuple[:6])
-    return datetime.datetime(*dtuple[:6],
-            tzinfo=datetime.timezone(datetime.timedelta(seconds=tz)))
+    try:
+        if tz is None:
+            return datetime.datetime(*dtuple[:6])
+        return datetime.datetime(*dtuple[:6],
+                tzinfo=datetime.timezone(datetime.timedelta(seconds=tz)))
+    except OverflowError as exc:
+        raise ValueError('Invalid date value or format "%s"' % str(data)) from exc
 
 
 def parseaddr(addr, *, strict=True):

Lib/html/parser.py

@@ -157,6 +157,9 @@ class HTMLParser(_markupbase.ParserBase):
         self.cdata_elem = None
         self._support_cdata = True
         self._escapable = True
+        self._pending = []
+        self._pending_len = 0
+        self._parse_threshold = 1
         super().reset()
 
     def feed(self, data):
@@ -165,11 +168,36 @@ class HTMLParser(_markupbase.ParserBase):
         Call this as often as you want, with as little or as much text
         as you want (may include '\n').
         """
-        self.rawdata = self.rawdata + data
-        self.goahead(0)
+        # Accumulate new data in a list and only join and parse it once
+        # enough has piled up.  Rescanning an unparsed buffer (e.g. an
+        # unterminated tag) and concatenating onto it on every call would
+        # both be quadratic in the input size.
+        self._pending_len += len(data)
+        if self._pending_len < self._parse_threshold:
+            self._pending.append(data)
+        else:
+            if not self._pending:
+                self.rawdata += data
+            else:
+                self._pending.append(data)
+                self.rawdata += ''.join(self._pending)
+                self._pending.clear()
+            self._pending_len = 0
+            n = len(self.rawdata)
+            self.goahead(0)
+            if len(self.rawdata) < n:
+                # Some data was parsed; resume on the next call.
+                self._parse_threshold = 1
+            else:
+                # Nothing was parsed; wait until the buffer doubles.
+                self._parse_threshold = len(self.rawdata)
 
     def close(self):
         """Handle any buffered data."""
+        if self._pending:
+            self.rawdata += ''.join(self._pending)
+            self._pending.clear()
+            self._pending_len = 0
         self.goahead(1)
 
     __starttag_text = None
@@ -387,9 +415,11 @@ class HTMLParser(_markupbase.ParserBase):
     def parse_comment(self, i, report=True):
         rawdata = self.rawdata
         assert rawdata.startswith('<!--', i), 'unexpected call to parse_comment()'
-        match = commentclose.search(rawdata, i+4)
+        # An empty comment is abruptly closed by the first ">" or "->",
+        # taking priority over a later "-->" or "--!>" close.
+        match = commentabruptclose.match(rawdata, i+4)
         if not match:
-            match = commentabruptclose.match(rawdata, i+4)
+            match = commentclose.search(rawdata, i+4)
             if not match:
                 return -1
         if report:

Lib/test/test_colorsys.py

@@ -42,6 +42,10 @@ class ColorsysTest(unittest.TestCase):
             self.assertTripleEqual(hsv, colorsys.rgb_to_hsv(*rgb))
             self.assertTripleEqual(rgb, colorsys.hsv_to_rgb(*hsv))
 
+            # test 360 phase shift in hue
+            h, s, v = hsv
+            self.assertTripleEqual(rgb, colorsys.hsv_to_rgb(h + 1.0, s, v))
+
     def test_hls_roundtrip(self):
         for r in frange(0.0, 1.0, 0.2):
             for g in frange(0.0, 1.0, 0.2):
@@ -89,6 +93,18 @@ class ColorsysTest(unittest.TestCase):
                         colorsys.yiq_to_rgb(*colorsys.rgb_to_yiq(*rgb))
                     )
 
+    def test_yiq_to_rgb_clamping(self):
+        values = [
+            # rgb, yiq (invalid YIQ values clamped to RGB range)
+            ((1.0, 0.0, 1.0), (0.0, 0.5, 1.0)),
+            ((0.0, 1.0, 0.0), (0.25, -1.0, -1.0)),
+            ((0.0, 0.0, 1.0), (0.0, -1.0, 0.5))
+        ]
+
+        for (rgb, yiq) in values:
+            with self.subTest(rgb=rgb, yiq=yiq):
+                self.assertTripleEqual(rgb, colorsys.yiq_to_rgb(*yiq))
+
     def test_yiq_values(self):
         values = [
             # rgb, yiq

Lib/test/test_email/test_headerregistry.py

@@ -223,6 +223,14 @@ class TestDateHeader(TestHeaderBase):
         self.assertEqual(len(h.defects), 1)
         self.assertIsInstance(h.defects[0], errors.InvalidDateDefect)
 
+    def test_out_of_range_date_value(self):
+        s = 'Mon, 20 Nov 9999999999 12:00:00 +0000'
+        h = self.make_header('date', s)
+        self.assertEqual(h, s)
+        self.assertIsNone(h.datetime)
+        self.assertEqual(len(h.defects), 1)
+        self.assertIsInstance(h.defects[0], errors.InvalidDateDefect)
+
     def test_datetime_read_only(self):
         h = self.make_header('date', self.datestring)
         with self.assertRaises(AttributeError):

Lib/test/test_email/test_utils.py

@@ -77,6 +77,15 @@ class DateTimeTests(unittest.TestCase):
             with self.subTest(dtstr=dtstr):
                 self.assertRaises(ValueError, utils.parsedate_to_datetime, dtstr)
 
+    def test_parsedate_to_datetime_out_of_range_raises_valueerror(self):
+        out_of_range_dates = [
+            'Mon, 20 Nov 9999999999 12:00:00 +0000',
+            'Mon, 20 Nov 2017 12:00:00 +24000000000000',
+        ]
+        for dtstr in out_of_range_dates:
+            with self.subTest(dtstr=dtstr):
+                self.assertRaises(ValueError, utils.parsedate_to_datetime, dtstr)
+
 class LocaltimeTests(unittest.TestCase):
 
     def test_localtime_is_tz_aware_daylight_true(self):

Lib/test/test_getopt.py

@@ -149,6 +149,18 @@ class GetoptTests(unittest.TestCase):
                                 ('-a', ''), ('--alpha', '')])
         self.assertEqual(args, ['arg1', 'arg2'])
 
+        # Allow string for single long argument
+        opts, args = getopt.getopt(cmdline, 'a::', 'alpha=?')
+        self.assertEqual(opts, [('-a', '1'), ('--alpha', '2'), ('--alpha', ''),
+                                ('-a', ''), ('--alpha', '')])
+        self.assertEqual(args, ['arg1', 'arg2'])
+
+        # Pass everything after -- as args
+        cmdline = ['-a1', '--alpha=2', '--', '-b', '--beta=5']
+        opts, args = getopt.getopt(cmdline, 'a:b', ['alpha=', 'beta'])
+        self.assertEqual(opts, [('-a', '1'), ('--alpha', '2')])
+        self.assertEqual(args, ['-b', '--beta=5'])
+
         self.assertError(getopt.getopt, cmdline, 'a:b', ['alpha', 'beta'])
 
     def test_gnu_getopt(self):
@@ -191,6 +203,25 @@ class GetoptTests(unittest.TestCase):
         self.assertEqual(args, ['arg1', '-b', '1', '--alpha', '--beta=2',
                                 '--beta', '3', 'arg2'])
 
+        # Allow string for single long argument
+        opts, args = getopt.gnu_getopt(cmdline, 'ab:', 'alpha')
+        self.assertEqual(opts, [('-a', '')])
+        self.assertEqual(args, ['arg1', '-b', '1', '--alpha', '--beta=2',
+                                '--beta', '3', 'arg2'])
+
+        # Pass everything after -- as args
+        cmdline = ['-a1', '--alpha=2', '--', '-b', '--beta=5']
+        opts, args = getopt.gnu_getopt(cmdline, 'a:b', ['alpha=', 'beta'])
+        self.assertEqual(opts, [('-a', '1'), ('--alpha', '2')])
+        self.assertEqual(args, ['-b', '--beta=5'])
+
+        # In order arguments
+        cmdline = ["gamma", "--alpha=3"]
+        opts, args = getopt.gnu_getopt(cmdline, '-', ["alpha="])
+        self.assertEqual(opts, [(None, ['gamma']), ('--alpha', '3')])
+        self.assertEqual(args, [])
+
+
     def test_issue4629(self):
         longopts, shortopts = getopt.getopt(['--help='], '', ['help='])
         self.assertEqual(longopts, [('--help', '')])
@@ -198,6 +229,10 @@ class GetoptTests(unittest.TestCase):
         self.assertEqual(longopts, [('--help', 'x')])
         self.assertRaises(getopt.GetoptError, getopt.getopt, ['--help='], '', ['help'])
 
+    def test_getopt_error_str(self):
+        error  = getopt.GetoptError('option -a not recognized', 'a')
+        self.assertEqual(str(error), 'option -a not recognized')
+
 def test_libref_examples():
     """
     Examples from the Library Reference:  Doc/lib/libgetopt.tex

Lib/test/test_htmlparser.py

@@ -116,6 +116,11 @@ class TestCaseBase(unittest.TestCase):
                    *, collector=None, convert_charrefs=False):
         if collector is None:
             collector = self.get_collector(convert_charrefs=convert_charrefs)
+            if isinstance(source, str):
+                # Also feed the whole string at once, not just character by
+                # character (below), to exercise different input buffering.
+                self._run_check([source], expected_events,
+                                convert_charrefs=convert_charrefs)
         parser = collector
         for s in source:
             parser.feed(s)
@@ -593,6 +598,9 @@ text
                 '<!-- <!-- nested --> -->'
                 '<!--<!-->'
                 '<!--<!--!>'
+                # abruptly closed empty comment must not swallow later text
+                '<!-->x-->'
+                '<!--->y-->'
         )
         expected = [('comment', " I'm a valid comment "),
                     ('comment', 'me too!'),
@@ -613,6 +621,8 @@ text
                     ('comment', ' <!-- nested '), ('data', ' -->'),
                     ('comment', '<!'),
                     ('comment', '<!'),
+                    ('comment', ''), ('data', 'x-->'),
+                    ('comment', ''), ('data', 'y-->'),
         ]
         self._run_check(html, expected)
 
@@ -1031,6 +1041,26 @@ text
         check("<![CDATA[" * 9 * n)
         check("<!doctype" * 35 * n)
 
+    @support.requires_resource('cpu')
+    def test_incremental_no_quadratic_complexity(self):
+        # An unterminated construct fed in many small chunks used to take
+        # quadratic time, both to rescan and to concatenate the buffer.
+        # Now it takes a fraction of a second.
+        def check(prefix, chunk, suffix):
+            parser = html.parser.HTMLParser()
+            parser.feed(prefix)
+            for _ in range(200_000):
+                parser.feed(chunk)
+            parser.feed(suffix)
+            parser.close()
+        chunk = "a" * 64
+        check("<!--", chunk, "-->")       # comment
+        check("<?", chunk, ">")           # processing instruction
+        check("<!doctype ", chunk, ">")   # doctype
+        check("<![CDATA[", chunk, "]]>")  # CDATA section
+        check("<a href='", chunk, "'>")   # start tag
+        check("<script>", chunk, "</script>")  # RAWTEXT element
+
 
 class AttributesTestCase(TestCaseBase):
 

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
 
@@ -368,6 +371,203 @@ class ShlexTest(unittest.TestCase):
         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'})

Lib/test/test_struct.py

@@ -181,6 +181,16 @@ class StructTest(ComplexesAreIdenticalMixin, unittest.TestCase):
         self.assertGreaterEqual(struct.calcsize('n'), struct.calcsize('i'))
         self.assertGreaterEqual(struct.calcsize('n'), struct.calcsize('P'))
 
+    def test_cache_bytes_vs_str_bb(self):
+        # Mixing str and bytes formats must not raise BytesWarning under -bb.
+        code = (
+            'import struct\n'
+            'struct.calcsize(b"!d"); struct.calcsize("!d")\n'
+            'struct.calcsize(">d"); struct.calcsize(b">d")\n'
+            'struct.Struct(b"i"); struct.Struct("i")\n'
+        )
+        assert_python_ok('-bb', '-c', code)
+
     def test_integers(self):
         # Integer tests (bBhHiIlLqQnN).
         import binascii