gh-153569: complete tokenizer validation coverage

Author: pablogsal

Lib/test/test_tokenize.py

@@ -2434,6 +2434,23 @@ def test_multiline_readline_chunk(self):
         ))
         self.assertEqual(tokens, expected)
 
+    def test_large_multiline_readline_chunk(self):
+        comment = "#" + "x" * (1024 * 1024)
+        lines = iter([(comment + "\npass\n").encode(), b""])
+        tokens = list(tokenize._generate_tokens_from_c_tokenizer(
+            lines.__next__, extra_tokens=True, encoding="utf-8"
+        ))
+        self.assertEqual(tokens[0].string, comment)
+        self.assertEqual(tokens[0].start, (1, 0))
+        self.assertEqual(tokens[0].end, (1, len(comment)))
+        self.assertEqual(
+            tokens[2],
+            tokenize.TokenInfo(
+                token.NAME, "pass", (2, 0), (2, 4), "pass\n"
+            ),
+        )
+        self.assertEqual(tokens[-1].type, token.ENDMARKER)
+
     def test_multiline_readline_chunk_with_unterminated_tail(self):
         readline = mock.Mock(side_effect=["x\nz", ""])
         iterator = _tokenize.TokenizerIter(readline, extra_tokens=True)
@@ -3874,6 +3891,12 @@ def test_newline_at_the_end_of_buffer(self):
             file_name = make_script(temp_dir, 'foo', test_script)
             run_test_script(file_name)
 
+    def test_large_file_reader_buffer(self):
+        test_script = "#" + "x" * (1024 * 1024) + "\npass\n"
+        with os_helper.temp_dir() as temp_dir:
+            file_name = make_script(temp_dir, "large_reader", test_script)
+            run_test_script(file_name)
+
 
 @support.force_not_colorized_test_class
 class CommandLineTest(unittest.TestCase):

Modules/_xxtestfuzz/dictionaries/fuzz_tokenizer.dict

@@ -0,0 +1,29 @@
+# delimiters and operators
+"("
+")"
+"["
+"]"
+"{"
+"}"
+":"
+"!"
+"="
+":="
+
+# strings and interpolation
+"f'"
+"f\""
+"t'"
+"t\""
+"'''"
+"\"\"\""
+"!r"
+"!s"
+"!a"
+
+# reader and decoder boundaries
+"# coding: utf-8\n"
+"# coding: latin-1\n"
+"\\\n"
+"\r\n"
+"\xc3\xa9"

Modules/_xxtestfuzz/fuzz_tokenizer_corpus/encoding_cookie.py

@@ -0,0 +1,2 @@
+A# coding: utf-8
+é = 1

Modules/_xxtestfuzz/fuzz_tokenizer_corpus/nested_fstring.py

@@ -0,0 +1 @@
+Gf'{value:{width}.{precision}f}'

Modules/_xxtestfuzz/fuzz_tokenizer_corpus/tstring.py

@@ -0,0 +1 @@
+Ct'{value=!r:{width}}'

Modules/_xxtestfuzz/fuzzer.c

@@ -594,6 +594,7 @@ static int fuzz_pycompile(const char* data, size_t size) {
 }
 
 #define MAX_TOKENIZER_TEST_SIZE 16384
+#define MAX_TOKENIZER_FILE_TEST_SIZE (2 * 1024 * 1024)
 
 static PyObject *tokenizer_iter_type;
 static PyObject *stringio_type;
@@ -717,11 +718,21 @@ fuzz_tokenizer_file(const char *data, size_t size, unsigned char options)
 static int
 fuzz_tokenizer(const char *data, size_t size)
 {
-    if (size < 1 || size > MAX_TOKENIZER_TEST_SIZE) {
+    if (size < 1) {
         return 0;
     }
 
     unsigned char options = (unsigned char)data[0];
+#ifdef FUZZ_TOKENIZER_FILE
+    size_t max_size = options & 0x04
+        ? MAX_TOKENIZER_FILE_TEST_SIZE
+        : MAX_TOKENIZER_TEST_SIZE;
+#else
+    size_t max_size = MAX_TOKENIZER_TEST_SIZE;
+#endif
+    if (size > max_size) {
+        return 0;
+    }
     data++;
     size--;
 

Tools/tokenizer/difftest.py

@@ -42,14 +42,35 @@ def positive_float(value):
 
 
 def exception_result(exc, *, locations=True):
+    message = getattr(exc, "msg", None)
+    token_location = None
+    if message is None and isinstance(exc, tokenize.TokenError):
+        if (
+            len(exc.args) == 2
+            and isinstance(exc.args[1], tuple)
+            and len(exc.args[1]) == 2
+        ):
+            message, token_location = exc.args
+        elif exc.args:
+            message = exc.args[0]
+    if message is None:
+        message = str(exc)
     result = {
         "type": type(exc).__name__,
-        "message": getattr(exc, "msg", str(exc)),
+        "message": message,
     }
     if locations:
         result.update({
-            "lineno": getattr(exc, "lineno", None),
-            "offset": getattr(exc, "offset", None),
+            "lineno": (
+                token_location[0]
+                if token_location is not None
+                else getattr(exc, "lineno", None)
+            ),
+            "offset": (
+                token_location[1]
+                if token_location is not None
+                else getattr(exc, "offset", None)
+            ),
             "end_lineno": getattr(exc, "end_lineno", None),
             "end_offset": getattr(exc, "end_offset", None),
             "text": getattr(exc, "text", None),
@@ -70,7 +91,9 @@ def warning_results(caught):
     ]
 
 
-def tokenizer_iter_result(data, extra_tokens, *, encoded=False, batched=False):
+def tokenizer_iter_result(
+    data, extra_tokens, *, encoded=False, batched=False, semantic=False
+):
     tokens = []
     with warnings.catch_warnings(record=True) as caught:
         warnings.simplefilter("always")
@@ -102,7 +125,7 @@ def readline():
             iterator = _tokenize.TokenizerIter(readline, **kwargs)
             tokens.extend(iterator)
         except Exception as exc:
-            error = exception_result(exc)
+            error = exception_result(exc, locations=not semantic)
         else:
             error = None
     return {
@@ -112,14 +135,14 @@ def readline():
     }
 
 
-def tokenize_result(data):
+def tokenize_result(data, semantic):
     tokens = []
     with warnings.catch_warnings(record=True) as caught:
         warnings.simplefilter("always")
         try:
             tokens.extend(tokenize.tokenize(io.BytesIO(data).readline))
         except Exception as exc:
-            error = exception_result(exc)
+            error = exception_result(exc, locations=not semantic)
         else:
             error = None
     return {
@@ -180,26 +203,34 @@ def case_result(case, batched, semantic):
         data = Path(case["path"]).read_bytes()
     else:
         data = base64.b64decode(case["data"], validate=True)
-    parser_tokens = tokenizer_iter_result(data, False)
-    extra_tokens = tokenizer_iter_result(data, True)
-    encoded_tokens = tokenizer_iter_result(data, False, encoded=True)
+    parser_tokens = tokenizer_iter_result(data, False, semantic=semantic)
+    extra_tokens = tokenizer_iter_result(data, True, semantic=semantic)
+    encoded_tokens = tokenizer_iter_result(
+        data, False, encoded=True, semantic=semantic
+    )
     return {
         "name": case["name"],
         "parser": compile_result(data, semantic),
-        "tokenize": tokenize_result(data),
+        "tokenize": tokenize_result(data, semantic),
         "parser_tokens": parser_tokens,
         "extra_tokens": extra_tokens,
         "encoded_tokens": encoded_tokens,
         "batched_tokens": (
-            tokenizer_iter_result(data, False, batched=True)
+            tokenizer_iter_result(
+                data, False, batched=True, semantic=semantic
+            )
             if batched else parser_tokens
         ),
         "batched_extra_tokens": (
-            tokenizer_iter_result(data, True, batched=True)
+            tokenizer_iter_result(
+                data, True, batched=True, semantic=semantic
+            )
             if batched else extra_tokens
         ),
         "batched_encoded_tokens": (
-            tokenizer_iter_result(data, False, encoded=True, batched=True)
+            tokenizer_iter_result(
+                data, False, encoded=True, batched=True, semantic=semantic
+            )
             if batched else encoded_tokens
         ),
     }