From 662beae6dec0b5b750a3c2730c934a5211e972bf Mon Sep 17 00:00:00 2001 From: Paul Ganssle Date: Thu, 22 Oct 2026 17:41:54 -0400 Subject: [PATCH 01/10] gh-107862: Add example.xfail() and example.via() to the Hypothesis stubs Hypothesis lets a test pin a known-failing input with `@example(...).xfail(reason=..., raises=...)`, which is the natural way to record a bug found by a property test without taking the input out of the strategy or writing a separate expectedFailure test. The stub `example` is a plain decorator function, so any test using `.xfail()` or `.via()` failed to import when Hypothesis is not installed. This turns the stub into a class with those two methods. An expected failure is strict, as in Hypothesis: when the examples are replayed the test must raise one of the `raises` exceptions for that input, and an unexpected pass fails the test. `.via()` is a no-op, as it is upstream. --- .../support/_hypothesis_stubs/__init__.py | 68 +++++++++++++++---- ...-10-22-10-00-00.gh-issue-107862.vX7pLm.rst | 3 + 2 files changed, 59 insertions(+), 12 deletions(-) create mode 100644 Misc/NEWS.d/next/Tests/2026-10-22-10-00-00.gh-issue-107862.vX7pLm.rst diff --git a/Lib/test/support/_hypothesis_stubs/__init__.py b/Lib/test/support/_hypothesis_stubs/__init__.py index e3cb54aad736d2..94b6f7eaf06a44 100644 --- a/Lib/test/support/_hypothesis_stubs/__init__.py +++ b/Lib/test/support/_hypothesis_stubs/__init__.py @@ -1,3 +1,4 @@ +from dataclasses import dataclass, replace from enum import Enum import functools import unittest @@ -18,21 +19,52 @@ from . import stateful, strategies +@dataclass(frozen=True) +class _Example: + args: tuple + kwargs: dict + # Set by example.xfail(): the exception types the example must raise. + raises: type[BaseException] | tuple[type[BaseException], ...] | None = None + reason: str = "" + + def __repr__(self): + bits = [repr(a) for a in self.args] + bits += [f"{k}={v!r}" for k, v in self.kwargs.items()] + return f"@example({', '.join(bits)})" + + +def _run_example(f, self, ex): + if ex.raises is None: + f(self, *ex.args, **ex.kwargs) + return + try: + f(self, *ex.args, **ex.kwargs) + except Exception as exc: + if not isinstance(exc, ex.raises): + raise + return + reason = f" because {ex.reason}" if ex.reason else "" + raise AssertionError( + f"Expected an exception from {ex!r}{reason}, " + "but no exception was raised" + ) + + def given(*_args, **_kwargs): def decorator(f): if examples := getattr(f, "_examples", []): @functools.wraps(f) def test_function(self): - for example_args, example_kwargs in examples: - if len(example_args) < 2: - subtest_args = example_args + for ex in examples: + if len(ex.args) < 2: + subtest_args = ex.args else: # subTest takes up to one positional argument. # When there are more, display them as a tuple - subtest_args = [example_args] - with self.subTest(*subtest_args, **example_kwargs): - f(self, *example_args, **example_kwargs) + subtest_args = [ex.args] + with self.subTest(*subtest_args, **ex.kwargs): + _run_example(f, self, ex) else: # If we have found no examples, we must skip the test. If @example @@ -49,16 +81,18 @@ def test_function(self): return decorator -def example(*args, **kwargs): - if bool(args) == bool(kwargs): - raise ValueError("Must specify exactly one of *args or **kwargs") +class example: + def __init__(self, *args, **kwargs): + if bool(args) == bool(kwargs): + raise ValueError("Must specify exactly one of *args or **kwargs") + self._example = _Example(args, kwargs) - def decorator(f): + def __call__(self, f): base_func = getattr(f, "__wrapped__", f) if not hasattr(base_func, "_examples"): base_func._examples = [] - base_func._examples.append((args, kwargs)) + base_func._examples.append(self._example) if getattr(f, "_given", False): # If the given decorator is below all the example decorators, @@ -68,7 +102,17 @@ def decorator(f): return f - return decorator + def xfail(self, condition=True, *, reason="", raises=BaseException): + # Like real Hypothesis, an expected failure is strict: the example + # must raise one of the given exceptions, or the test fails. + if condition: + self._example = replace(self._example, raises=raises, + reason=reason) + return self + + def via(self, whence, /): + # A label for tooling; it does not change what runs. + return self def assume(condition): diff --git a/Misc/NEWS.d/next/Tests/2026-10-22-10-00-00.gh-issue-107862.vX7pLm.rst b/Misc/NEWS.d/next/Tests/2026-10-22-10-00-00.gh-issue-107862.vX7pLm.rst new file mode 100644 index 00000000000000..6c583e75c39c56 --- /dev/null +++ b/Misc/NEWS.d/next/Tests/2026-10-22-10-00-00.gh-issue-107862.vX7pLm.rst @@ -0,0 +1,3 @@ +Add ``example.xfail()`` and ``example.via()`` to the Hypothesis stubs used +when Hypothesis is not installed, so that property tests can mark an explicit +example as an expected failure in both modes. From d8af5abb237d0c6f1a93dc637754cef4475b475a Mon Sep 17 00:00:00 2001 From: Paul Ganssle Date: Wed, 21 Oct 2026 20:04:44 -0400 Subject: [PATCH 02/10] gh-107862: Add Hypothesis property tests for the codecs machinery The existing codec tests check each codec against fixed strings, and test_basics runs the incremental and stream interfaces on "abc123" only. This adds test_codecs_property.py, which runs every codec in all_unicode_encodings through the incremental codecs (any chunking, getstate/setstate round trips), StreamReader/StreamWriter, the error handlers (backslashreplace decoded back through unicode_escape, xmlcharrefreplace through html.unescape, surrogateescape round trips), charmap_decode/charmap_encode against a small oracle, name normalization in codecs.lookup, and the transform codecs, all on generated input and compared against the stateless codec functions. Six divergences turned up, each recorded as an expectedFailure with the minimal input as an explicit example and carved out of the broad tests only where the comments say so: - euc_jp, euc_jis_2004, euc_jisx0213 and gb18030 incremental decoders drop input after an incomplete trailing sequence under surrogateescape. - The ISO-2022 encoders emit U+001B as a bare ESC (iso2022_kr also SO and SI) which their decoders reject or misread. - StreamReader.read() never decodes with final=True, so an incomplete trailing character is silently dropped. - The punycode incremental codecs transform each chunk on its own. - euc_kr encodes U+3164 to the Hangul filler bytes it cannot decode alone. - iso2022_jp_3 decodes plane 2 0x7D3B with the 2004 table because the 2000 emulation is followed by "if" rather than "else if". The by-design differences (BOM-less utf-16/32 incrementally, utf-7 and lone surrogates, the many-to-one compatibility mappings in cp932, cp950, euc_jp and shift_jis, surrogateescape on non-ASCII-compatible codecs) are documented in the module rather than hidden in the strategies. With Hypothesis the file runs in about five seconds; 3000 examples per test pass as well. Without it the explicit examples replay through the stubs. --- Lib/test/test_codecs_property.py | 822 ++++++++++++++++++ ...-10-21-20-30-00.gh-issue-107862.kQ3zT7.rst | 3 + 2 files changed, 825 insertions(+) create mode 100644 Lib/test/test_codecs_property.py create mode 100644 Misc/NEWS.d/next/Tests/2026-10-21-20-30-00.gh-issue-107862.kQ3zT7.rst diff --git a/Lib/test/test_codecs_property.py b/Lib/test/test_codecs_property.py new file mode 100644 index 00000000000000..10d41c957bee54 --- /dev/null +++ b/Lib/test/test_codecs_property.py @@ -0,0 +1,822 @@ +"""Property tests for the codec machinery. + +These run every registered text codec through the incremental, stream and +error-handler interfaces on generated input and compare against the stateless +one-shot codec functions, which the other codec tests treat as the reference. +With Hypothesis installed the input is generated; without it, only the +explicit examples run. +""" + +import codecs +import encodings +import html +import io +import unittest + +from test.support.hypothesis_helper import hypothesis +from test.test_codecs import all_unicode_encodings, bytes_transform_encodings + +st = hypothesis.strategies + + +def _available(encoding): + try: + codecs.lookup(encoding) + except LookupError: + return False + return True + + +# The Windows-only codecs only support a few error handlers. +UNICODE_ENCODINGS = tuple(e for e in all_unicode_encodings + if e not in ("mbcs", "oem") and _available(e)) + +# These transform a whole label at a time, so their incremental and stream +# variants do not accept arbitrary chunking, and idna supports 'strict' only. +WHOLE_STRING_ENCODINGS = frozenset({"idna", "punycode"}) + +# The stateless decoder assumes native byte order when there is no byte +# order mark; the incremental decoder refuses the input instead. +BOM_ENCODINGS = frozenset({"utf_16", "utf_32"}) + +# Implemented by Modules/cjkcodecs; they share one decoder loop. +MULTIBYTE_ENCODINGS = frozenset( + e for e in UNICODE_ENCODINGS + if codecs.lookup(e).incrementaldecoder.__module__ == f"encodings.{e}" + and "_multibytecodec" in repr(codecs.lookup(e).incrementaldecoder.__mro__)) + +ISO2022_ENCODINGS = frozenset(e for e in UNICODE_ENCODINGS + if e.startswith("iso2022")) + +# The multibyte codecs with sequences longer than two bytes; see +# MultibyteSurrogateescapeBug. +LONG_SEQUENCE_ENCODINGS = frozenset( + {"euc_jp", "euc_jis_2004", "euc_jisx0213", "gb18030"}) + +INCREMENTAL_ENCODINGS = tuple(e for e in UNICODE_ENCODINGS + if e not in WHOLE_STRING_ENCODINGS) + +DECODE_ERRORS = ("strict", "ignore", "replace", "backslashreplace", + "surrogateescape") +ENCODE_ERRORS = DECODE_ERRORS + ("xmlcharrefreplace", "namereplace", + "surrogatepass") + +# Encoders that are many-to-one by design, so decode(encode(s)) need not +# give s back: idna applies nameprep, raw_unicode_escape leaves existing +# backslashes alone (it exists for pickle, not for round trips), and the +# Japanese and Big5 codecs fold a few compatibility characters (U+00A5 and +# U+203E onto ASCII bytes, U+00A2 onto U+FFE0 and so on). +NON_INJECTIVE_ENCODINGS = frozenset( + {"idna", "raw_unicode_escape", "cp932", "cp950", "euc_jp", "shift_jis"}) + +# Characters an encoder accepts although its decoder cannot give them back; +# each is pinned under expectedFailure in RoundTripBugs and kept out of the +# generated text until fixed. +ENCODE_ONLY_CHARACTERS = { + # A bare ESC, which the decoders take as the start of a sequence, and + # for iso2022_kr the SO and SI shift characters as well. + **dict.fromkeys(ISO2022_ENCODINGS, "\x1b"), + "iso2022_kr": "\x1b\x0e\x0f", + # The Hangul filler, which the decoder only accepts inside a sequence. + "euc_kr": "\u3164", + # Encoded with the JIS X 0213:2000 table but decoded with the 2004 one. + "iso2022_jp_3": "\u9b1d", +} + + +def _lossless_bytes(encoding): + # Whether every single byte survives decode(surrogateescape) followed + # by encode(surrogateescape). That rules out codecs with undefined + # bytes below 0x80 (surrogateescape refuses to escape those), tables + # mapping two bytes to one character, and the byte-order-mark codecs. + for b in range(256): + data = bytes([b]) + try: + if data.decode(encoding, "surrogateescape").encode( + encoding, "surrogateescape") != data: + return False + except UnicodeError: + return False + return True + + +# surrogateescape round trips wherever decoding is injective on bytes and +# the codec is a superset of ASCII. Multibyte codecs can have duplicate +# two-byte sequences, and the escape codecs rewrite backslashes. +SURROGATEESCAPE_ENCODINGS = tuple( + e for e in UNICODE_ENCODINGS + if e not in MULTIBYTE_ENCODINGS | WHOLE_STRING_ENCODINGS + and e not in ("unicode_escape", "raw_unicode_escape", "utf_7") + and _lossless_bytes(e)) + + +text = st.text(max_size=64) +# st.text() leaves surrogates out; the error handlers are mostly about them. +surrogates = st.characters(categories=["Cs"]) +text_with_surrogates = st.text( + alphabet=st.characters() | surrogates, max_size=64) +data = st.binary(max_size=64) +sizes = st.lists(st.integers(0, 8), max_size=16) +incremental_encoding = st.sampled_from(INCREMENTAL_ENCODINGS) + + +def _chunks(buf, sizes): + # Split buf at the given sizes; the remainder is the last chunk. + out = [] + pos = 0 + for n in sizes: + out.append(buf[pos:pos + n]) + pos += n + out.append(buf[pos:]) + return out + + +def _outcome(func): + # Normalize a call to something comparable: its result, or the type + # of the UnicodeError it raised. + try: + return ("ok", func()) + except UnicodeError as exc: + return ("error", type(exc)) + + +def _with_bom(encoding, data, big_endian): + # Give the byte-order-mark codecs a mark, so that the stateless and + # incremental decoders agree on the byte order (see + # test_bom_codecs_reject_bare_input for what happens without one). + if encoding in BOM_ENCODINGS: + suffix = "_be" if big_endian else "_le" + return "\ufeff".encode(encoding + suffix) + data + return data + + +def _decode_errors(encoding): + # These decoders drop input after an incomplete trailing sequence + # under surrogateescape; pinned by MultibyteSurrogateescapeBug. + if encoding in LONG_SEQUENCE_ENCODINGS: + return tuple(e for e in DECODE_ERRORS if e != "surrogateescape") + return DECODE_ERRORS + + +def _encode_text(encoding): + # utf-7 encodes a lone surrogate instead of rejecting it, and joins + # a surrogate pair into one code point, so a pair split across chunks + # encodes differently from the pair in one piece. + if encoding == "utf_7": + return text + return text_with_surrogates + + +def _round_trip_text(encoding, strategy=text): + # Text that the codec can encode and decode again, see + # ENCODE_ONLY_CHARACTERS. + bad = ENCODE_ONLY_CHARACTERS.get(encoding) + if bad: + table = str.maketrans("", "", bad) + return strategy.map(lambda s: s.translate(table)) + return strategy + + +decode_case = incremental_encoding.flatmap( + lambda e: st.tuples(st.just(e), st.sampled_from(_decode_errors(e)))) +encode_case = incremental_encoding.flatmap( + lambda e: st.tuples(st.just(e), st.sampled_from(ENCODE_ERRORS), + _encode_text(e))) +round_trip_case = st.sampled_from(UNICODE_ENCODINGS).flatmap( + lambda e: st.tuples(st.just(e), _round_trip_text(e))) +# For the tests that run the error handlers' output back through the +# codec: idna accepts no handler but 'strict', and the many-to-one codecs +# do not give the input back. +handler_encoding = st.sampled_from(tuple( + e for e in UNICODE_ENCODINGS + if e != "idna" and e not in NON_INJECTIVE_ENCODINGS)) + + +class IncrementalCodecTests(unittest.TestCase): + """The incremental codecs agree with the stateless ones for any + chunking of the input, including under the error handlers.""" + + @hypothesis.given(case=decode_case, data=data, sizes=sizes, + big_endian=st.booleans()) + @hypothesis.example(case=("utf_8", "strict"), data=b"", sizes=[], + big_endian=False) + @hypothesis.example(case=("utf_8", "replace"), data=b"a\xe2\x82\xac\xe2", + sizes=[2, 1, 1], big_endian=False) + @hypothesis.example(case=("utf_16", "strict"), data=b"a\x00\xd8", + sizes=[1, 1], big_endian=True) + @hypothesis.example(case=("utf_7", "strict"), data=b"+AOk-+", sizes=[3], + big_endian=False) + @hypothesis.example(case=("euc_jp", "replace"), data=b"\xa4\xa2\xa4", + sizes=[1, 1], big_endian=False) + @hypothesis.example(case=("iso2022_jp", "ignore"), + data=b"\x1b$B$\"\x1b(B", sizes=[1, 2, 1], + big_endian=False) + @hypothesis.example(case=("cp1252", "backslashreplace"), data=b"\x81a", + sizes=[1], big_endian=False) + @hypothesis.example(case=("unicode_escape", "strict"), data=b"\\u00e9\\", + sizes=[3, 2], big_endian=False) + def test_decode_in_chunks(self, case, data, sizes, big_endian): + encoding, errors = case + data = _with_bom(encoding, data, big_endian) + expected = _outcome(lambda: data.decode(encoding, errors)) + + def incremental(): + decoder = codecs.getincrementaldecoder(encoding)(errors) + parts = [decoder.decode(chunk) for chunk in _chunks(data, sizes)] + parts.append(decoder.decode(b"", final=True)) + return "".join(parts) + + self.assertEqual(_outcome(incremental), expected) + + @hypothesis.given(case=encode_case, sizes=sizes) + @hypothesis.example(case=("utf_8", "strict", ""), sizes=[]) + @hypothesis.example(case=("utf_8", "surrogatepass", "a\ud800b"), + sizes=[1, 1]) + @hypothesis.example(case=("utf_16", "strict", "ab"), sizes=[1]) + @hypothesis.example(case=("utf_7", "strict", "a\xe9b\xe9"), sizes=[2, 1]) + @hypothesis.example(case=("iso2022_jp", "replace", "a\u3042\xe9b"), + sizes=[1, 1, 1]) + @hypothesis.example(case=("hz", "xmlcharrefreplace", "~\u4e2d\u20ac"), + sizes=[1, 1]) + @hypothesis.example(case=("ascii", "namereplace", "\xe9"), sizes=[]) + @hypothesis.example(case=("cp437", "ignore", "a\u4e2db"), sizes=[2]) + def test_encode_in_chunks(self, case, sizes): + encoding, errors, text = case + expected = _outcome(lambda: text.encode(encoding, errors)) + + def incremental(): + encoder = codecs.getincrementalencoder(encoding)(errors) + parts = [encoder.encode(chunk) for chunk in _chunks(text, sizes)] + parts.append(encoder.encode("", final=True)) + return b"".join(parts) + + actual = _outcome(incremental) + if encoding == "utf_7" and actual[0] == expected[0] == "ok": + # The encoder closes its shift sequence at each chunk end, so + # the bytes differ but must decode to the same text. + actual = actual[1].decode(encoding) + expected = expected[1].decode(encoding) + self.assertEqual(actual, expected) + + @hypothesis.given(encoding=incremental_encoding, text=text, sizes=sizes) + @hypothesis.example(encoding="utf_8", text="", sizes=[]) + @hypothesis.example(encoding="utf_16", text="a\u20acb", sizes=[1, 1]) + @hypothesis.example(encoding="iso2022_kr", text="\uac00a\uac01", + sizes=[1, 1]) + @hypothesis.example(encoding="cp037", text="a\xe9", sizes=[1]) + def test_iterencode_iterdecode(self, encoding, text, sizes): + try: + expected = text.encode(encoding).decode(encoding) + except UnicodeError: + return + chunks = _chunks(text, sizes) + encoded = list(codecs.iterencode(chunks, encoding)) + self.assertTrue(all(isinstance(e, bytes) for e in encoded)) + decoded = "".join(codecs.iterdecode(encoded, encoding)) + self.assertEqual(decoded, expected) + + @hypothesis.given(case=decode_case, data=data, split=st.integers(0, 64), + big_endian=st.booleans()) + @hypothesis.example(case=("utf_8", "strict"), data=b"", split=0, + big_endian=False) + @hypothesis.example(case=("utf_8", "strict"), data=b"a\xe2\x82\xac", + split=2, big_endian=False) + @hypothesis.example(case=("utf_16", "replace"), data=b"a\x00\x00\xd8b", + split=3, big_endian=False) + @hypothesis.example(case=("utf_7", "strict"), data=b"a+AOk-b", split=3, + big_endian=False) + @hypothesis.example(case=("gb18030", "replace"), + data=b"\x81\x30\x81\x30a", split=2, big_endian=False) + @hypothesis.example(case=("iso2022_jp", "strict"), + data=b"\x1b$B$\"\x1b(Ba", split=4, big_endian=False) + @hypothesis.example(case=("unicode_escape", "strict"), data=b"a\\x41", + split=2, big_endian=False) + def test_decoder_state_round_trip(self, case, data, split, big_endian): + # The conditions documented for IncrementalDecoder.getstate(). + encoding, errors = case + data = _with_bom(encoding, data, big_endian) + split = min(split, len(data)) + decoder = codecs.getincrementaldecoder(encoding)(errors) + try: + head = decoder.decode(data[:split]) + except UnicodeError: + return + state = decoder.getstate() + self.assertIsInstance(state[0], bytes) + self.assertIsInstance(state[1], int) + if not state[1]: + # Nothing but buffered bytes: feeding them to a fresh decoder + # gives no output and reproduces the state. + decoder.setstate((b"", 0)) + self.assertEqual(decoder.decode(state[0]), "") + self.assertEqual(decoder.getstate(), state) + + resumed = codecs.getincrementaldecoder(encoding)(errors) + resumed.setstate(state) + actual = _outcome(lambda: head + resumed.decode(data[split:], True)) + expected = _outcome(lambda: data.decode(encoding, errors)) + self.assertEqual(actual, expected) + + @hypothesis.given(case=encode_case, split=st.integers(0, 64)) + @hypothesis.example(case=("utf_8", "strict", ""), split=0) + @hypothesis.example(case=("utf_16", "strict", "ab"), split=1) + @hypothesis.example(case=("utf_7", "strict", "a\xe9\xe9b"), split=2) + @hypothesis.example(case=("iso2022_jp", "strict", "\u3042\u3044a"), + split=1) + @hypothesis.example(case=("hz", "replace", "\u4e2d\u20ac\u4e2d"), + split=2) + @hypothesis.example(case=("latin_1", "surrogateescape", "a\udc80b"), + split=1) + def test_encoder_state_round_trip(self, case, split): + encoding, errors, text = case + split = min(split, len(text)) + encoder = codecs.getincrementalencoder(encoding)(errors) + try: + head = encoder.encode(text[:split]) + except UnicodeError: + return + state = encoder.getstate() + resumed = codecs.getincrementalencoder(encoding)(errors) + resumed.setstate(state) + actual = _outcome(lambda: head + resumed.encode(text[split:], True)) + expected = _outcome(lambda: text.encode(encoding, errors)) + if encoding == "utf_7" and actual[0] == expected[0] == "ok": + actual = actual[1].decode(encoding) + expected = expected[1].decode(encoding) + self.assertEqual(actual, expected) + + @hypothesis.given(encoding=st.sampled_from(sorted(BOM_ENCODINGS)), + data=data) + @hypothesis.example(encoding="utf_16", data=b"") + @hypothesis.example(encoding="utf_16", data=b"a\x00") + @hypothesis.example(encoding="utf_16", data=b"\xff\xfea\x00") + @hypothesis.example(encoding="utf_32", data=b"a\x00\x00") + @hypothesis.example(encoding="utf_32", data=b"\x00\x00\xfe\xffa") + def test_bom_codecs_reject_bare_input(self, encoding, data): + # Without a byte order mark the incremental decoder raises, where + # the stateless decoder would assume native order. + decoder = codecs.getincrementaldecoder(encoding)() + boms = tuple("\ufeff".encode(encoding + suffix) + for suffix in ("_le", "_be")) + if data.startswith(boms): + self.assertEqual(_outcome(lambda: decoder.decode(data, True)), + _outcome(lambda: data.decode(encoding))) + elif not data: + self.assertEqual(decoder.decode(data, True), "") + else: + with self.assertRaises(UnicodeDecodeError): + decoder.decode(data, final=True) + + +class RoundTripTests(unittest.TestCase): + + @hypothesis.given(case=round_trip_case) + @hypothesis.example(case=("utf_8", "")) + @hypothesis.example(case=("utf_8", "a\u20ac\U0001f600")) + @hypothesis.example(case=("utf_7", "a+b\u20ac")) + @hypothesis.example(case=("shift_jis", "\xa5~")) + @hypothesis.example(case=("iso2022_jp_2", "\u3042\xe9\uac00a")) + @hypothesis.example(case=("big5hkscs", "\u00ca\u0304")) + @hypothesis.example(case=("punycode", "b\xfccher")) + @hypothesis.example(case=("idna", "b\xfccher.example")) + @hypothesis.example(case=("unicode_escape", "\\n\n\u20ac")) + def test_encode_decode(self, case): + encoding, text = case + try: + encoded = text.encode(encoding) + except UnicodeEncodeError: + return + self.assertIsInstance(encoded, bytes) + decoded = encoded.decode(encoding) + # encode is always a left inverse of decode on its own output... + self.assertEqual(decoded.encode(encoding), encoded) + # ...and a full inverse unless it is many-to-one by design. + if encoding not in NON_INJECTIVE_ENCODINGS: + self.assertEqual(decoded, text) + + @hypothesis.given(encoding=st.sampled_from(SURROGATEESCAPE_ENCODINGS), + data=data) + @hypothesis.example(encoding="utf_8", data=b"") + @hypothesis.example(encoding="utf_8", data=b"a\xff\xe2\x82") + @hypothesis.example(encoding="ascii", data=b"\x80\xff") + @hypothesis.example(encoding="cp1252", data=b"\x81\x8d") + def test_surrogateescape_round_trip(self, encoding, data): + decoded = data.decode(encoding, "surrogateescape") + self.assertEqual(decoded.encode(encoding, "surrogateescape"), data) + # The escapes are exactly the undecodable bytes. + escaped = sum(1 for c in decoded if "\udc80" <= c <= "\udcff") + try: + data.decode(encoding) + except UnicodeDecodeError: + self.assertGreater(escaped, 0) + else: + self.assertEqual(escaped, 0) + + @hypothesis.given(case=handler_encoding.flatmap( + lambda e: st.tuples(st.just(e), _round_trip_text( + e, text_with_surrogates.map(lambda s: s.replace("\\", "")))))) + @hypothesis.example(case=("ascii", "")) + @hypothesis.example(case=("ascii", "a\xe9\u20ac\U0001f600\udc80")) + @hypothesis.example(case=("cp1252", "\u20ac\u0100")) + @hypothesis.example(case=("euc_kr", "\u3042\u20ac")) + @hypothesis.example(case=("iso2022_jp_2004", "\xa0\u20ac")) + @hypothesis.example(case=("punycode", "\xfc\udc80")) + def test_backslashreplace_decodes_back(self, case): + # backslashreplace writes unencodable characters as escapes that + # unicode_escape understands, so as long as the text has no + # backslashes of its own the original comes back. + encoding, text = case + encoded = text.encode(encoding, "backslashreplace") + decoded = encoded.decode(encoding) + as_ascii = decoded.encode("ascii", "backslashreplace") + self.assertEqual(as_ascii.decode("unicode_escape"), text) + + @hypothesis.given(case=handler_encoding.flatmap( + # html.unescape maps "&" sequences and treats the C1 controls as + # cp1252, so keep both out of the text. + lambda e: st.tuples(st.just(e), _round_trip_text(e, text.map( + lambda s: "".join(c for c in s + if c != "&" and not "\x80" <= c <= "\x9f")))))) + @hypothesis.example(case=("ascii", "")) + @hypothesis.example(case=("ascii", "a<\xe9>\u20ac\U0001f600")) + @hypothesis.example(case=("utf_7", "+\u20ac-")) + @hypothesis.example(case=("shift_jis_2004", "\u3042\u20ac")) + @hypothesis.example(case=("iso2022_kr", "\uac00\u20ac")) + def test_xmlcharrefreplace_unescapes_back(self, case): + encoding, text = case + encoded = text.encode(encoding, "xmlcharrefreplace") + self.assertEqual(html.unescape(encoded.decode(encoding)), text) + + @hypothesis.given(data=data) + @hypothesis.example(data=b"") + @hypothesis.example(data=b"\\\x00\n'\"\xff") + def test_escape_codec(self, data): + escaped, consumed = codecs.escape_encode(data) + self.assertEqual(consumed, len(data)) + self.assertTrue(escaped.isascii()) + self.assertEqual(codecs.escape_decode(escaped), (data, len(escaped))) + + @hypothesis.given(text=text_with_surrogates) + @hypothesis.example(text="") + @hypothesis.example(text="\\\n\x00\xe9\u20ac\ud800\U0001f600") + def test_unicode_escape_with_surrogates(self, text): + for encoding in ("unicode_escape", "raw_unicode_escape"): + with self.subTest(encoding=encoding): + encoded = text.encode(encoding) + self.assertEqual(encoded.decode(encoding), + text.encode(encoding).decode(encoding)) + encoded = text.encode("unicode_escape") + self.assertTrue(encoded.isascii()) + self.assertEqual(encoded.decode("unicode_escape"), text) + + @hypothesis.given(encoding=st.sampled_from(bytes_transform_encodings), + data=data) + @hypothesis.example(encoding="base64_codec", data=b"") + @hypothesis.example(encoding="quopri_codec", data=b"=\r\n \xff") + @hypothesis.example(encoding="uu_codec", data=b"\x00" * 45 + b"a") + def test_bytes_transform(self, encoding, data): + encoded = codecs.encode(data, encoding) + self.assertIsInstance(encoded, bytes) + self.assertEqual(codecs.decode(encoded, encoding), data) + + @hypothesis.given(text=text) + @hypothesis.example(text="") + @hypothesis.example(text="Hello, \u20ac!") + def test_rot13(self, text): + encoded = codecs.encode(text, "rot_13") + self.assertEqual(len(encoded), len(text)) + self.assertEqual(codecs.decode(encoded, "rot_13"), text) + self.assertEqual(codecs.encode(encoded, "rot_13"), text) + + +UNDEFINED = "\ufffe" +decoding_table = st.lists( + st.characters() | st.just(UNDEFINED), min_size=256, max_size=256, +).map("".join) + + +def _charmap_oracle(data, errors, table): + # What charmap_decode should produce, from the documented behavior of + # the error handlers on a one-byte-at-a-time codec. + out = [] + for i, b in enumerate(data): + c = table[b] + if c != UNDEFINED: + out.append(c) + elif errors == "ignore": + pass + elif errors == "replace": + out.append("\ufffd") + elif errors == "backslashreplace": + out.append(f"\\x{b:02x}") + elif errors == "surrogateescape" and b >= 0x80: + out.append(chr(0xdc00 + b)) + else: + raise UnicodeDecodeError("charmap", data, i, i + 1, "undefined") + return "".join(out) + + +class CharmapTests(unittest.TestCase): + + @hypothesis.given(table=decoding_table, data=data, + errors=st.sampled_from(DECODE_ERRORS)) + @hypothesis.example(table="".join(map(chr, range(256))), data=b"", + errors="strict") + @hypothesis.example(table=UNDEFINED * 128 + "".join(map(chr, range(128))), + data=b"\x00\x80\xff", errors="replace") + @hypothesis.example(table=UNDEFINED * 128 + "".join(map(chr, range(128))), + data=b"\x00\x80\xff", errors="surrogateescape") + @hypothesis.example(table=UNDEFINED * 256, data=b"\x00", errors="strict") + def test_decode_matches_table(self, table, data, errors): + expected = _outcome(lambda: _charmap_oracle(data, errors, table)) + actual = _outcome(lambda: codecs.charmap_decode(data, errors, table)) + if actual[0] == "ok": + actual = ("ok", actual[1][0]) + self.assertEqual( + codecs.charmap_decode(data, errors, table)[1], len(data)) + self.assertEqual(actual, expected) + # A dict mapping is accepted too and must agree. + mapping = {i: c for i, c in enumerate(table) if c != UNDEFINED} + self.assertEqual( + _outcome(lambda: codecs.charmap_decode(data, errors, mapping)), + _outcome(lambda: codecs.charmap_decode(data, errors, table))) + + @hypothesis.given( + table=st.lists(st.characters(), min_size=256, max_size=256, + unique=True).map("".join), + indices=st.lists(st.integers(0, 255), max_size=32), + errors=st.sampled_from(ENCODE_ERRORS)) + @hypothesis.example(table="".join(map(chr, range(256))), indices=[], + errors="strict") + @hypothesis.example(table="".join(map(chr, range(256, 512))), + indices=[0, 255, 1], errors="strict") + def test_encode_inverts_table(self, table, indices, errors): + # With an injective table, encoding characters from the table gives + # back their positions, through both kinds of encoding map. + text = "".join(table[i] for i in indices) + expected = bytes(indices) + built = codecs.charmap_build(table) + mapping = {ord(c): i for i, c in enumerate(table)} + self.assertEqual(codecs.charmap_encode(text, errors, built), + (expected, len(text))) + self.assertEqual(codecs.charmap_encode(text, errors, mapping), + (expected, len(text))) + self.assertEqual(codecs.charmap_decode(expected, "strict", table), + (text, len(expected))) + + @hypothesis.given( + table=st.lists(st.characters(), min_size=256, max_size=256, + unique=True).map("".join), + text=text_with_surrogates, + errors=st.sampled_from(ENCODE_ERRORS)) + @hypothesis.example(table="".join(map(chr, range(256))), text="\u20ac", + errors="strict") + @hypothesis.example(table="".join(map(chr, range(256))), + text="a\u20ac\udc80", errors="xmlcharrefreplace") + @hypothesis.example(table="".join(map(chr, range(256))), + text="a\u20ac\udc80", errors="surrogateescape") + def test_encode_maps_agree(self, table, text, errors): + # The EncodingMap from charmap_build and an equivalent dict give + # the same result or raise the same error. + built = codecs.charmap_build(table) + mapping = {ord(c): i for i, c in enumerate(table)} + self.assertEqual( + _outcome(lambda: codecs.charmap_encode(text, errors, built)), + _outcome(lambda: codecs.charmap_encode(text, errors, mapping))) + + +# Characters that str.splitlines() treats as line boundaries, plus filler. +line_text = st.text( + alphabet=st.sampled_from("ab \n\r\x0b\x0c\x1c\x1d\x1e\x85\u2028\u2029"), + max_size=64) + + +class StreamTests(unittest.TestCase): + + @hypothesis.given(encoding=incremental_encoding, text=text, + write_sizes=sizes, + read_sizes=st.lists(st.integers(-1, 8), max_size=16)) + @hypothesis.example(encoding="utf_8", text="", write_sizes=[], + read_sizes=[]) + @hypothesis.example(encoding="utf_8", text="a\u20ac\U0001f600", + write_sizes=[1, 1], read_sizes=[1, 2, -1]) + @hypothesis.example(encoding="utf_16", text="a\u20acb", write_sizes=[1], + read_sizes=[2, 0, 1]) + @hypothesis.example(encoding="utf_7", text="a\xe9\xe9b", write_sizes=[2], + read_sizes=[3]) + @hypothesis.example(encoding="iso2022_jp", text="\u3042a\u3044", + write_sizes=[1, 1], read_sizes=[1, 1, 1]) + def test_writer_then_reader(self, encoding, text, write_sizes, + read_sizes): + try: + expected = text.encode(encoding).decode(encoding) + except UnicodeError: + return + stream = io.BytesIO() + writer = codecs.getwriter(encoding)(stream) + for chunk in _chunks(text, write_sizes): + writer.write(chunk) + self.assertEqual(stream.getvalue().decode(encoding), expected) + + stream.seek(0) + reader = codecs.getreader(encoding)(stream) + parts = [reader.read(n) for n in read_sizes] + for n, part in zip(read_sizes, parts): + if n >= 0: + self.assertLessEqual(len(part), n) + parts.append(reader.read()) + self.assertEqual("".join(parts), expected) + self.assertEqual(reader.read(), "") + + @hypothesis.given(text=line_text, keepends=st.booleans()) + @hypothesis.example(text="", keepends=True) + @hypothesis.example(text="a\r\nb\rc\nd", keepends=True) + @hypothesis.example(text="a\r\nb\rc\nd", keepends=False) + @hypothesis.example(text="\u2028\x85\n\r", keepends=True) + def test_reader_readline(self, text, keepends): + expected = text.splitlines(keepends) + reader = codecs.getreader("utf_8")(io.BytesIO(text.encode())) + lines = [] + while line := reader.readline(keepends=keepends): + lines.append(line) + if not keepends: + # An empty line without its ending is indistinguishable from + # the end of the stream, so the loop above stops early there. + n = len(lines) + self.assertEqual(lines, expected[:n]) + self.assertTrue(n == len(expected) or expected[n] == "") + else: + self.assertEqual(lines, expected) + + reader = codecs.getreader("utf_8")(io.BytesIO(text.encode())) + self.assertEqual(reader.readlines(keepends=keepends), expected) + reader = codecs.getreader("utf_8")(io.BytesIO(text.encode())) + self.assertEqual(list(reader), text.splitlines(True)) + + @hypothesis.given( + text=line_text, + ops=st.lists(st.integers(-1, 6) | st.just("line") | st.just("next"), + max_size=16)) + @hypothesis.example(text="", ops=[]) + @hypothesis.example(text="a\r\nb\rc\nd\n", ops=["line", 1, "line", -1]) + @hypothesis.example(text="a\nb\nc\nd\n", + ops=["line", "line", 0, 2, "next"]) + @hypothesis.example(text="ab\r\n", ops=[3, "line", "line"]) + def test_reader_mixed_reads(self, text, ops): + # Interleaving read() and readline() exercises the reader's line + # buffer; the result must match reading from the decoded text. + reader = codecs.getreader("utf_8")(io.BytesIO(text.encode())) + remaining = text + for op in ops: + if op == "line" or op == "next": + lines = remaining.splitlines(True) + expected = lines[0] if lines else "" + if op == "next" and not expected: + with self.assertRaises(StopIteration): + next(reader) + continue + actual = next(reader) if op == "next" else reader.readline() + elif op < 0: + expected = remaining + actual = reader.read() + else: + expected = remaining[:op] + actual = reader.read(op) + self.assertEqual(actual, expected) + remaining = remaining[len(expected):] + self.assertEqual(reader.read(), remaining) + + +class LookupTests(unittest.TestCase): + + @hypothesis.given( + encoding=st.sampled_from(UNICODE_ENCODINGS + ("rot_13",)), + separators=st.lists(st.sampled_from(["_", "-", " ", "--", "_ -"]), + min_size=8, max_size=8), + upper=st.lists(st.booleans(), min_size=16, max_size=16)) + @hypothesis.example(encoding="utf_8", separators=["-"] * 8, + upper=[True] * 16) + @hypothesis.example(encoding="iso8859_1", separators=["_ -"] * 8, + upper=[False, True] * 8) + @hypothesis.example(encoding="shift_jis_2004", separators=[" "] * 8, + upper=[True] * 16) + def test_lookup_name_normalization(self, encoding, separators, upper): + # Case and the punctuation between name parts do not matter. + parts = encoding.split("_") + joined = "".join( + part + (separators[i] if i < len(parts) - 1 else "") + for i, part in enumerate(parts)) + variant = "".join(c.upper() if upper[i % len(upper)] else c + for i, c in enumerate(joined)) + self.assertEqual(codecs.lookup(variant).name, + codecs.lookup(encoding).name) + self.assertEqual(encodings.normalize_encoding(variant).lower(), + encodings.normalize_encoding(encoding)) + if encoding != "rot_13": + self.assertEqual("x".encode(variant), "x".encode(encoding)) + + +class MultibyteSurrogateescapeBug(unittest.TestCase): + """An incomplete sequence of three or four bytes at the end of the + input makes the incremental decoder drop whatever follows the first + escaped byte, where the stateless decoder escapes that byte and carries + on. Carved out of the chunked decoding tests above by _decode_errors.""" + + @unittest.expectedFailure + @hypothesis.given( + encoding=st.sampled_from(sorted(LONG_SEQUENCE_ENCODINGS)), + data=data) + @hypothesis.example(encoding="gb18030", data=b"\x80" b"6") + @hypothesis.example(encoding="euc_jp", data=b"\x8f\x00") + @hypothesis.example(encoding="euc_jis_2004", data=b"\xa8\x8fl") + def test_trailing_bytes_after_escape(self, encoding, data): + expected = _outcome(lambda: data.decode(encoding, "surrogateescape")) + decoder = codecs.getincrementaldecoder(encoding)("surrogateescape") + self.assertEqual(_outcome(lambda: decoder.decode(data, True)), + expected) + + +class RoundTripBugs(unittest.TestCase): + """Characters an encoder accepts but its decoder cannot give back; see + ENCODE_ONLY_CHARACTERS for how the round-trip tests avoid them.""" + + def _check_round_trip(self, encoding, text): + try: + encoded = text.encode(encoding) + except UnicodeEncodeError: + return + self.assertEqual(encoded.decode(encoding), text) + + # The ISO-2022 encoders pass U+001B through as a bare ESC byte, which + # their own decoders then reject or misread as an escape sequence; + # iso2022_kr does the same with its shift characters U+000E and U+000F. + @unittest.expectedFailure + @hypothesis.given(encoding=st.sampled_from(sorted(ISO2022_ENCODINGS)), + text=text) + @hypothesis.example(encoding="iso2022_jp", text="\x1b") + @hypothesis.example(encoding="iso2022_jp_2004", text="\x1b\xa0") + @hypothesis.example(encoding="iso2022_kr", text="a\x1b\uac00") + @hypothesis.example(encoding="iso2022_kr", text="\x0e") + @hypothesis.example(encoding="iso2022_kr", text="a\x0f\uac00") + def test_iso2022_escape(self, encoding, text): + self._check_round_trip(encoding, text) + + # euc_kr encodes the Hangul filler U+3164 as the two bytes that start + # an eight-byte composed syllable, and will not decode them alone. + @unittest.expectedFailure + @hypothesis.given(text=text) + @hypothesis.example(text="\u3164") + @hypothesis.example(text="a\u3164\uac00") + def test_euc_kr_filler(self, text): + self._check_round_trip("euc_kr", text) + + # iso2022_jp_3 encodes with the JIS X 0213:2000 table but decodes with + # the 2004 one, where the same code point means U+9B1C. + @unittest.expectedFailure + @hypothesis.given(text=text) + @hypothesis.example(text="\u9b1d") + @hypothesis.example(text="a\u9b1d\u3042") + def test_iso2022_jp_3_mapping(self, text): + self._check_round_trip("iso2022_jp_3", text) + + +class StreamReaderTruncationBug(unittest.TestCase): + """StreamReader.read() never tells the decoder that the stream has + ended, so an incomplete trailing sequence is silently dropped instead + of raising (or being replaced), as it is everywhere else.""" + + @unittest.expectedFailure + @hypothesis.given( + encoding=st.sampled_from(["utf_8", "utf_16_le", "utf_32_be", + "utf_16", "raw_unicode_escape"]), + text=text, errors=st.sampled_from(["strict", "replace"])) + @hypothesis.example(encoding="utf_8", text="", errors="strict") + @hypothesis.example(encoding="utf_8", text="a", errors="replace") + @hypothesis.example(encoding="utf_16", text="\u3042", errors="strict") + def test_read_reports_truncated_input(self, encoding, text, errors): + # Chop the last byte off a character that takes more than one. + encoded = (text + "\u3042").encode(encoding)[:-1] + expected = _outcome(lambda: encoded.decode(encoding, errors)) + reader = codecs.getreader(encoding)(io.BytesIO(encoded), errors) + self.assertEqual(_outcome(lambda: reader.read() + reader.read()), + expected) + + +class PunycodeIncrementalBug(unittest.TestCase): + """The punycode incremental codecs encode and decode each chunk on its + own, so anything fed in more than one piece comes out wrong. Punycode + is left out of the chunked tests above as a whole-string codec.""" + + @unittest.expectedFailure + @hypothesis.given(text=text, sizes=sizes) + @hypothesis.example(text="ab", sizes=[1]) + @hypothesis.example(text="b\xfccher", sizes=[2, 2]) + def test_chunked_encode(self, text, sizes): + encoder = codecs.getincrementalencoder("punycode")() + parts = [encoder.encode(chunk) for chunk in _chunks(text, sizes)] + parts.append(encoder.encode("", final=True)) + self.assertEqual(b"".join(parts), text.encode("punycode")) + + +if __name__ == "__main__": + unittest.main() diff --git a/Misc/NEWS.d/next/Tests/2026-10-21-20-30-00.gh-issue-107862.kQ3zT7.rst b/Misc/NEWS.d/next/Tests/2026-10-21-20-30-00.gh-issue-107862.kQ3zT7.rst new file mode 100644 index 00000000000000..995d8f99c38b4d --- /dev/null +++ b/Misc/NEWS.d/next/Tests/2026-10-21-20-30-00.gh-issue-107862.kQ3zT7.rst @@ -0,0 +1,3 @@ +Add Hypothesis property tests for the :mod:`codecs` machinery: the +incremental, stream and error-handler interfaces of every text codec are +compared against the stateless codec functions on generated input. From 3ca2eb2a5893860bba7fddec2612c42650c36c1f Mon Sep 17 00:00:00 2001 From: Paul Ganssle Date: Wed, 21 Oct 2026 20:16:26 -0400 Subject: [PATCH 03/10] gh-107862: Reference the existing issues for two of the codec expected failures The StreamReader truncation is gh-56717, open since 2011 with the same diagnosis, and the euc_kr filler asymmetry was closed as not a bug in gh-101863; the punycode chunking is mentioned in gh-64331. --- Lib/test/test_codecs_property.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/Lib/test/test_codecs_property.py b/Lib/test/test_codecs_property.py index 10d41c957bee54..de0887ce349a36 100644 --- a/Lib/test/test_codecs_property.py +++ b/Lib/test/test_codecs_property.py @@ -762,7 +762,8 @@ def test_iso2022_escape(self, encoding, text): self._check_round_trip(encoding, text) # euc_kr encodes the Hangul filler U+3164 as the two bytes that start - # an eight-byte composed syllable, and will not decode them alone. + # an eight-byte composed syllable, and will not decode them alone + # (gh-101863, closed as not a bug; recorded here as the asymmetry it is). @unittest.expectedFailure @hypothesis.given(text=text) @hypothesis.example(text="\u3164") @@ -783,7 +784,7 @@ def test_iso2022_jp_3_mapping(self, text): class StreamReaderTruncationBug(unittest.TestCase): """StreamReader.read() never tells the decoder that the stream has ended, so an incomplete trailing sequence is silently dropped instead - of raising (or being replaced), as it is everywhere else.""" + of raising (or being replaced), as it is everywhere else (gh-56717).""" @unittest.expectedFailure @hypothesis.given( @@ -804,8 +805,9 @@ def test_read_reports_truncated_input(self, encoding, text, errors): class PunycodeIncrementalBug(unittest.TestCase): """The punycode incremental codecs encode and decode each chunk on its - own, so anything fed in more than one piece comes out wrong. Punycode - is left out of the chunked tests above as a whole-string codec.""" + own, so anything fed in more than one piece comes out wrong (noted in + passing in gh-64331). Punycode is left out of the chunked tests above + as a whole-string codec.""" @unittest.expectedFailure @hypothesis.given(text=text, sizes=sizes) From 05ee77afc4e4cbfa6041ed2e5baf55fdf81515b2 Mon Sep 17 00:00:00 2001 From: Paul Ganssle Date: Wed, 21 Oct 2026 20:27:34 -0400 Subject: [PATCH 04/10] gh-107862: Generate only failing inputs in the codec expected failures Previously the expectedFailure tests drew from the same broad strategies as the properties they were carved out of, so most of their generated inputs passed and only the explicit examples made them fail. Each one now generates just the known-failing shape (a trailing incomplete three- or four-byte sequence, a trailing ESC/SO/SI, a filler or the 0x7D3B character between printable ASCII, a chopped multi-byte character, two non-empty ASCII punycode chunks), so the whole test flips to passing once the bug is fixed. A search for a passing input in each strategy finds none. --- Lib/test/test_codecs_property.py | 102 ++++++++++++++++++++----------- 1 file changed, 65 insertions(+), 37 deletions(-) diff --git a/Lib/test/test_codecs_property.py b/Lib/test/test_codecs_property.py index de0887ce349a36..713ed9044dbe71 100644 --- a/Lib/test/test_codecs_property.py +++ b/Lib/test/test_codecs_property.py @@ -716,6 +716,24 @@ def test_lookup_name_normalization(self, encoding, separators, upper): self.assertEqual("x".encode(variant), "x".encode(encoding)) +# The expected failures below each generate only inputs that are known to +# fail, so that they start passing as a whole once the bug is fixed. + +# Printable ASCII, which every codec below encodes to itself. +printable_ascii = st.characters(min_codepoint=0x20, max_codepoint=0x7e) +ascii_text = st.text(alphabet=printable_ascii, max_size=32) +ascii_bytes = st.binary(max_size=32).map(lambda b: bytes(c & 0x7f for c in b)) + + +def _incomplete_long_sequence(encoding): + # The first two bytes of a three- or four-byte sequence, where the + # second byte is ASCII: the stateless decoder escapes the first byte + # and decodes the second as itself. + if encoding == "gb18030": + return st.tuples(st.integers(0x81, 0xfe), st.integers(0x30, 0x39)) + return st.tuples(st.just(0x8f), st.integers(0, 0x7f)) + + class MultibyteSurrogateescapeBug(unittest.TestCase): """An incomplete sequence of three or four bytes at the end of the input makes the incremental decoder drop whatever follows the first @@ -724,12 +742,15 @@ class MultibyteSurrogateescapeBug(unittest.TestCase): @unittest.expectedFailure @hypothesis.given( - encoding=st.sampled_from(sorted(LONG_SEQUENCE_ENCODINGS)), - data=data) - @hypothesis.example(encoding="gb18030", data=b"\x80" b"6") - @hypothesis.example(encoding="euc_jp", data=b"\x8f\x00") - @hypothesis.example(encoding="euc_jis_2004", data=b"\xa8\x8fl") - def test_trailing_bytes_after_escape(self, encoding, data): + case=st.sampled_from(sorted(LONG_SEQUENCE_ENCODINGS)).flatmap( + lambda e: st.tuples(st.just(e), ascii_bytes, + _incomplete_long_sequence(e)))) + @hypothesis.example(case=("gb18030", b"", (0x80, ord("6")))) + @hypothesis.example(case=("euc_jp", b"a", (0x8f, 0))) + @hypothesis.example(case=("euc_jis_2004", b"", (0x8f, ord("l")))) + def test_trailing_bytes_after_escape(self, case): + encoding, prefix, tail = case + data = prefix + bytes(tail) expected = _outcome(lambda: data.decode(encoding, "surrogateescape")) decoder = codecs.getincrementaldecoder(encoding)("surrogateescape") self.assertEqual(_outcome(lambda: decoder.decode(data, True)), @@ -741,44 +762,45 @@ class RoundTripBugs(unittest.TestCase): ENCODE_ONLY_CHARACTERS for how the round-trip tests avoid them.""" def _check_round_trip(self, encoding, text): - try: - encoded = text.encode(encoding) - except UnicodeEncodeError: - return + encoded = text.encode(encoding) self.assertEqual(encoded.decode(encoding), text) # The ISO-2022 encoders pass U+001B through as a bare ESC byte, which # their own decoders then reject or misread as an escape sequence; # iso2022_kr does the same with its shift characters U+000E and U+000F. + # The control character goes last, where it never forms a sequence. @unittest.expectedFailure - @hypothesis.given(encoding=st.sampled_from(sorted(ISO2022_ENCODINGS)), - text=text) - @hypothesis.example(encoding="iso2022_jp", text="\x1b") - @hypothesis.example(encoding="iso2022_jp_2004", text="\x1b\xa0") - @hypothesis.example(encoding="iso2022_kr", text="a\x1b\uac00") - @hypothesis.example(encoding="iso2022_kr", text="\x0e") - @hypothesis.example(encoding="iso2022_kr", text="a\x0f\uac00") - def test_iso2022_escape(self, encoding, text): - self._check_round_trip(encoding, text) + @hypothesis.given( + case=st.sampled_from(sorted(ISO2022_ENCODINGS)).flatmap( + lambda e: st.tuples( + st.just(e), ascii_text, + st.sampled_from(ENCODE_ONLY_CHARACTERS[e])))) + @hypothesis.example(case=("iso2022_jp", "", "\x1b")) + @hypothesis.example(case=("iso2022_jp_2004", "a", "\x1b")) + @hypothesis.example(case=("iso2022_kr", "", "\x0e")) + @hypothesis.example(case=("iso2022_kr", "a", "\x0f")) + def test_iso2022_escape(self, case): + encoding, prefix, control = case + self._check_round_trip(encoding, prefix + control) # euc_kr encodes the Hangul filler U+3164 as the two bytes that start # an eight-byte composed syllable, and will not decode them alone # (gh-101863, closed as not a bug; recorded here as the asymmetry it is). @unittest.expectedFailure - @hypothesis.given(text=text) - @hypothesis.example(text="\u3164") - @hypothesis.example(text="a\u3164\uac00") - def test_euc_kr_filler(self, text): - self._check_round_trip("euc_kr", text) + @hypothesis.given(prefix=ascii_text, suffix=ascii_text) + @hypothesis.example(prefix="", suffix="") + @hypothesis.example(prefix="a", suffix="b") + def test_euc_kr_filler(self, prefix, suffix): + self._check_round_trip("euc_kr", prefix + "\u3164" + suffix) # iso2022_jp_3 encodes with the JIS X 0213:2000 table but decodes with # the 2004 one, where the same code point means U+9B1C. @unittest.expectedFailure - @hypothesis.given(text=text) - @hypothesis.example(text="\u9b1d") - @hypothesis.example(text="a\u9b1d\u3042") - def test_iso2022_jp_3_mapping(self, text): - self._check_round_trip("iso2022_jp_3", text) + @hypothesis.given(prefix=ascii_text, suffix=ascii_text) + @hypothesis.example(prefix="", suffix="") + @hypothesis.example(prefix="a", suffix="b") + def test_iso2022_jp_3_mapping(self, prefix, suffix): + self._check_round_trip("iso2022_jp_3", prefix + "\u9b1d" + suffix) class StreamReaderTruncationBug(unittest.TestCase): @@ -790,7 +812,10 @@ class StreamReaderTruncationBug(unittest.TestCase): @hypothesis.given( encoding=st.sampled_from(["utf_8", "utf_16_le", "utf_32_be", "utf_16", "raw_unicode_escape"]), - text=text, errors=st.sampled_from(["strict", "replace"])) + # No backslashes: raw_unicode_escape leaves them alone, and an even + # number in front of the chopped escape turns it into literal text. + text=text.map(lambda s: s.replace("\\", "")), + errors=st.sampled_from(["strict", "replace"])) @hypothesis.example(encoding="utf_8", text="", errors="strict") @hypothesis.example(encoding="utf_8", text="a", errors="replace") @hypothesis.example(encoding="utf_16", text="\u3042", errors="strict") @@ -809,15 +834,18 @@ class PunycodeIncrementalBug(unittest.TestCase): passing in gh-64331). Punycode is left out of the chunked tests above as a whole-string codec.""" + # Two non-empty ASCII chunks give two "-" terminated labels where the + # whole string gives one. @unittest.expectedFailure - @hypothesis.given(text=text, sizes=sizes) - @hypothesis.example(text="ab", sizes=[1]) - @hypothesis.example(text="b\xfccher", sizes=[2, 2]) - def test_chunked_encode(self, text, sizes): + @hypothesis.given( + head=st.text(alphabet=printable_ascii, min_size=1, max_size=32), + tail=st.text(alphabet=printable_ascii, min_size=1, max_size=32)) + @hypothesis.example(head="a", tail="b") + @hypothesis.example(head="b\xfc", tail="cher") + def test_chunked_encode(self, head, tail): encoder = codecs.getincrementalencoder("punycode")() - parts = [encoder.encode(chunk) for chunk in _chunks(text, sizes)] - parts.append(encoder.encode("", final=True)) - self.assertEqual(b"".join(parts), text.encode("punycode")) + chunked = encoder.encode(head) + encoder.encode(tail, final=True) + self.assertEqual(chunked, (head + tail).encode("punycode")) if __name__ == "__main__": From dd857cafa967db1cafcef965cb14a65f3ceca61b Mon Sep 17 00:00:00 2001 From: Paul Ganssle Date: Thu, 22 Oct 2026 14:47:51 -0400 Subject: [PATCH 05/10] gh-107862: Reference gh-160302 from the ISO-2022 expected failure --- Lib/test/test_codecs_property.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/Lib/test/test_codecs_property.py b/Lib/test/test_codecs_property.py index 713ed9044dbe71..392a02a3752257 100644 --- a/Lib/test/test_codecs_property.py +++ b/Lib/test/test_codecs_property.py @@ -767,8 +767,9 @@ def _check_round_trip(self, encoding, text): # The ISO-2022 encoders pass U+001B through as a bare ESC byte, which # their own decoders then reject or misread as an escape sequence; - # iso2022_kr does the same with its shift characters U+000E and U+000F. - # The control character goes last, where it never forms a sequence. + # iso2022_kr does the same with its shift characters U+000E and U+000F + # (gh-160302). The control character goes last, where it never forms + # a sequence. @unittest.expectedFailure @hypothesis.given( case=st.sampled_from(sorted(ISO2022_ENCODINGS)).flatmap( From 65304abfa0a06963879b3ea27a5f588d3b2dbc17 Mon Sep 17 00:00:00 2001 From: Paul Ganssle Date: Thu, 22 Oct 2026 14:50:47 -0400 Subject: [PATCH 06/10] gh-107862: Reference gh-160306 from the iso2022_jp_3 expected failure --- Lib/test/test_codecs_property.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Lib/test/test_codecs_property.py b/Lib/test/test_codecs_property.py index 392a02a3752257..fdcff204451725 100644 --- a/Lib/test/test_codecs_property.py +++ b/Lib/test/test_codecs_property.py @@ -795,7 +795,7 @@ def test_euc_kr_filler(self, prefix, suffix): self._check_round_trip("euc_kr", prefix + "\u3164" + suffix) # iso2022_jp_3 encodes with the JIS X 0213:2000 table but decodes with - # the 2004 one, where the same code point means U+9B1C. + # the 2004 one, where the same code point means U+9B1C (gh-160306). @unittest.expectedFailure @hypothesis.given(prefix=ascii_text, suffix=ascii_text) @hypothesis.example(prefix="", suffix="") From b59fbde4ccabcbd60afb3f41f034fe56c41da7e0 Mon Sep 17 00:00:00 2001 From: Paul Ganssle Date: Thu, 22 Oct 2026 14:52:33 -0400 Subject: [PATCH 07/10] gh-107862: Reference gh-160307 from the surrogateescape expected failure --- Lib/test/test_codecs_property.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/Lib/test/test_codecs_property.py b/Lib/test/test_codecs_property.py index fdcff204451725..8d3e6891a1187f 100644 --- a/Lib/test/test_codecs_property.py +++ b/Lib/test/test_codecs_property.py @@ -738,7 +738,8 @@ class MultibyteSurrogateescapeBug(unittest.TestCase): """An incomplete sequence of three or four bytes at the end of the input makes the incremental decoder drop whatever follows the first escaped byte, where the stateless decoder escapes that byte and carries - on. Carved out of the chunked decoding tests above by _decode_errors.""" + on (gh-160307). Carved out of the chunked decoding tests above by + _decode_errors.""" @unittest.expectedFailure @hypothesis.given( From baa4e0bbcf0e564ee06b3db0738ba204a3cffbde Mon Sep 17 00:00:00 2001 From: Paul Ganssle Date: Thu, 22 Oct 2026 14:54:05 -0400 Subject: [PATCH 08/10] gh-107862: Reference gh-160308 from the punycode expected failure --- Lib/test/test_codecs_property.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/Lib/test/test_codecs_property.py b/Lib/test/test_codecs_property.py index 8d3e6891a1187f..fbac40976a01c7 100644 --- a/Lib/test/test_codecs_property.py +++ b/Lib/test/test_codecs_property.py @@ -832,9 +832,9 @@ def test_read_reports_truncated_input(self, encoding, text, errors): class PunycodeIncrementalBug(unittest.TestCase): """The punycode incremental codecs encode and decode each chunk on its - own, so anything fed in more than one piece comes out wrong (noted in - passing in gh-64331). Punycode is left out of the chunked tests above - as a whole-string codec.""" + own, so anything fed in more than one piece comes out wrong + (gh-160308). Punycode is left out of the chunked tests above as a + whole-string codec.""" # Two non-empty ASCII chunks give two "-" terminated labels where the # whole string gives one. From 087a37eb8ee7a03e74227f6e9f3beabf14dad2b3 Mon Sep 17 00:00:00 2001 From: Paul Ganssle Date: Thu, 22 Oct 2026 14:55:06 -0400 Subject: [PATCH 09/10] gh-107862: Reference gh-160309 from the StreamReader expected failure --- Lib/test/test_codecs_property.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/Lib/test/test_codecs_property.py b/Lib/test/test_codecs_property.py index fbac40976a01c7..183fc3679ba4e9 100644 --- a/Lib/test/test_codecs_property.py +++ b/Lib/test/test_codecs_property.py @@ -808,7 +808,8 @@ def test_iso2022_jp_3_mapping(self, prefix, suffix): class StreamReaderTruncationBug(unittest.TestCase): """StreamReader.read() never tells the decoder that the stream has ended, so an incomplete trailing sequence is silently dropped instead - of raising (or being replaced), as it is everywhere else (gh-56717).""" + of raising (or being replaced), as it is everywhere else (gh-160309, + the same mechanism as gh-56717).""" @unittest.expectedFailure @hypothesis.given( From ab96b89734fc05a765cb507015869c831014067c Mon Sep 17 00:00:00 2001 From: Paul Ganssle Date: Thu, 22 Oct 2026 17:54:05 -0400 Subject: [PATCH 10/10] gh-107862: Pin the known codec failures as expected-failure examples Previously each known bug had its own expectedFailure test generating only the failing inputs. Hypothesis already has a way to say that: an `@example(...).xfail(reason=..., raises=...)` on the test the input belongs to, which is strict in both Hypothesis and (after gh-160356) the stubs. Each bug is now a few such examples on test_decode_in_chunks, test_encode_in_chunks, test_encode_decode or test_reader_matches_decode, and the carve-outs in the strategies point at them. Also from review: - MULTIBYTE_ENCODINGS is an issubclass check against _multibytecodec.MultibyteIncrementalDecoder rather than a look at the repr of the MRO. - Only raw_unicode_escape is listed out of the surrogateescape codecs by hand, and for the right reason (a truncated \u escape raises); unicode_escape and utf_7 already fail _lossless_bytes. - xmlcharrefreplace text leaves out the control characters and noncharacters whose references html.unescape drops, not only "&" and the C1 range. - The charmap encoding tables leave out U+FFFE, which charmap_build treats as undefined, and the two encode tests share the strategy. - test_unicode_escape_with_surrogates compared an expression with itself; it now checks the unicode_escape round trip and the raw_unicode_escape round trip on text without backslashes. - StreamReader.read(size) takes a byte count, so the reads are no longer expected to return at most that many characters (big5hkscs decodes one sequence to two). test_reader_read_chars covers the chars argument of the pure Python reader, which the multibyte readers do not have, and test_reader_matches_decode compares the reader with the stateless decoder under every error handler. --- Lib/test/test_codecs_property.py | 381 +++++++++++++++++-------------- 1 file changed, 204 insertions(+), 177 deletions(-) diff --git a/Lib/test/test_codecs_property.py b/Lib/test/test_codecs_property.py index 183fc3679ba4e9..0ee46b96d13d9c 100644 --- a/Lib/test/test_codecs_property.py +++ b/Lib/test/test_codecs_property.py @@ -5,6 +5,10 @@ one-shot codec functions, which the other codec tests treat as the reference. With Hypothesis installed the input is generated; without it, only the explicit examples run. + +Inputs that are known to fail are pinned as expected-failure examples on +the test they belong to, with the issue number as the reason, and are kept +out of the generated input until the bug is fixed. """ import codecs @@ -16,6 +20,12 @@ from test.support.hypothesis_helper import hypothesis from test.test_codecs import all_unicode_encodings, bytes_transform_encodings +try: + import _multibytecodec +except ImportError: + # Then none of the CJK codecs is available either. + _multibytecodec = None + st = hypothesis.strategies @@ -39,23 +49,30 @@ def _available(encoding): # order mark; the incremental decoder refuses the input instead. BOM_ENCODINGS = frozenset({"utf_16", "utf_32"}) -# Implemented by Modules/cjkcodecs; they share one decoder loop. +# Implemented by Modules/cjkcodecs; they share one decoder loop and have +# their own stream reader and writer. MULTIBYTE_ENCODINGS = frozenset( e for e in UNICODE_ENCODINGS - if codecs.lookup(e).incrementaldecoder.__module__ == f"encodings.{e}" - and "_multibytecodec" in repr(codecs.lookup(e).incrementaldecoder.__mro__)) + if _multibytecodec is not None + and issubclass(codecs.lookup(e).incrementaldecoder, + _multibytecodec.MultibyteIncrementalDecoder)) ISO2022_ENCODINGS = frozenset(e for e in UNICODE_ENCODINGS if e.startswith("iso2022")) -# The multibyte codecs with sequences longer than two bytes; see -# MultibyteSurrogateescapeBug. +# The multibyte codecs with sequences longer than two bytes; see the +# expected-failure examples on test_decode_in_chunks (gh-160307). LONG_SEQUENCE_ENCODINGS = frozenset( {"euc_jp", "euc_jis_2004", "euc_jisx0213", "gb18030"}) INCREMENTAL_ENCODINGS = tuple(e for e in UNICODE_ENCODINGS if e not in WHOLE_STRING_ENCODINGS) +# The codecs whose StreamReader is the pure Python one from the codecs +# module, as opposed to the multibyte codecs' own. +PYTHON_READER_ENCODINGS = tuple(e for e in INCREMENTAL_ENCODINGS + if e not in MULTIBYTE_ENCODINGS) + DECODE_ERRORS = ("strict", "ignore", "replace", "backslashreplace", "surrogateescape") ENCODE_ERRORS = DECODE_ERRORS + ("xmlcharrefreplace", "namereplace", @@ -70,16 +87,18 @@ def _available(encoding): {"idna", "raw_unicode_escape", "cp932", "cp950", "euc_jp", "shift_jis"}) # Characters an encoder accepts although its decoder cannot give them back; -# each is pinned under expectedFailure in RoundTripBugs and kept out of the -# generated text until fixed. +# each is pinned as an expected-failure example on test_encode_decode and +# kept out of the generated text until fixed. ENCODE_ONLY_CHARACTERS = { # A bare ESC, which the decoders take as the start of a sequence, and - # for iso2022_kr the SO and SI shift characters as well. + # for iso2022_kr the SO and SI shift characters as well (gh-160302). **dict.fromkeys(ISO2022_ENCODINGS, "\x1b"), "iso2022_kr": "\x1b\x0e\x0f", - # The Hangul filler, which the decoder only accepts inside a sequence. + # The Hangul filler, which the decoder only accepts inside a sequence + # (gh-101863). "euc_kr": "\u3164", - # Encoded with the JIS X 0213:2000 table but decoded with the 2004 one. + # Encoded with the JIS X 0213:2000 table but decoded with the 2004 one + # (gh-160306). "iso2022_jp_3": "\u9b1d", } @@ -102,12 +121,12 @@ def _lossless_bytes(encoding): # surrogateescape round trips wherever decoding is injective on bytes and # the codec is a superset of ASCII. Multibyte codecs can have duplicate -# two-byte sequences, and the escape codecs rewrite backslashes. +# two-byte sequences, and raw_unicode_escape raises on a truncated \u +# escape, whose bytes are ASCII and so cannot be escaped either. SURROGATEESCAPE_ENCODINGS = tuple( e for e in UNICODE_ENCODINGS if e not in MULTIBYTE_ENCODINGS | WHOLE_STRING_ENCODINGS - and e not in ("unicode_escape", "raw_unicode_escape", "utf_7") - and _lossless_bytes(e)) + and e != "raw_unicode_escape" and _lossless_bytes(e)) text = st.text(max_size=64) @@ -117,6 +136,7 @@ def _lossless_bytes(encoding): alphabet=st.characters() | surrogates, max_size=64) data = st.binary(max_size=64) sizes = st.lists(st.integers(0, 8), max_size=16) +read_sizes = st.lists(st.integers(-1, 8), max_size=16) incremental_encoding = st.sampled_from(INCREMENTAL_ENCODINGS) @@ -152,7 +172,7 @@ def _with_bom(encoding, data, big_endian): def _decode_errors(encoding): # These decoders drop input after an incomplete trailing sequence - # under surrogateescape; pinned by MultibyteSurrogateescapeBug. + # under surrogateescape (gh-160307); pinned on test_decode_in_chunks. if encoding in LONG_SEQUENCE_ENCODINGS: return tuple(e for e in DECODE_ERRORS if e != "surrogateescape") return DECODE_ERRORS @@ -215,6 +235,30 @@ class IncrementalCodecTests(unittest.TestCase): sizes=[1], big_endian=False) @hypothesis.example(case=("unicode_escape", "strict"), data=b"\\u00e9\\", sizes=[3, 2], big_endian=False) + # gh-160307: after escaping the first byte of an incomplete three- or + # four-byte sequence at the end of the input, these decoders drop the + # bytes that follow it. The EUC codecs do this for 0x8f followed by + # any ASCII byte, gb18030 for any byte above 0x7f followed by a digit. + @hypothesis.example(case=("gb18030", "surrogateescape"), data=b"\x806", + sizes=[], big_endian=False).xfail( + reason="gh-160307", raises=AssertionError) + @hypothesis.example(case=("gb18030", "surrogateescape"), data=b"a\x819", + sizes=[1], big_endian=False).xfail( + reason="gh-160307", raises=AssertionError) + @hypothesis.example(case=("euc_jp", "surrogateescape"), data=b"a\x8f\x00", + sizes=[1], big_endian=False).xfail( + reason="gh-160307", raises=AssertionError) + @hypothesis.example(case=("euc_jis_2004", "surrogateescape"), + data=b"\x8fl", sizes=[], big_endian=False).xfail( + reason="gh-160307", raises=AssertionError) + @hypothesis.example(case=("euc_jisx0213", "surrogateescape"), + data=b"\x8fA", sizes=[1], big_endian=False).xfail( + reason="gh-160307", raises=AssertionError) + # gh-160308: the punycode incremental decoder decodes each chunk on + # its own. + @hypothesis.example(case=("punycode", "strict"), data=b"ab-", sizes=[1], + big_endian=False).xfail( + reason="gh-160308", raises=AssertionError) def test_decode_in_chunks(self, case, data, sizes, big_endian): encoding, errors = case data = _with_bom(encoding, data, big_endian) @@ -240,6 +284,13 @@ def incremental(): sizes=[1, 1]) @hypothesis.example(case=("ascii", "namereplace", "\xe9"), sizes=[]) @hypothesis.example(case=("cp437", "ignore", "a\u4e2db"), sizes=[2]) + # gh-160308: the punycode incremental encoder encodes each chunk on + # its own, so two chunks give two "-" terminated labels. + @hypothesis.example(case=("punycode", "strict", "ab"), sizes=[1]).xfail( + reason="gh-160308", raises=AssertionError) + @hypothesis.example(case=("punycode", "strict", "b\xfccher"), + sizes=[2]).xfail( + reason="gh-160308", raises=AssertionError) def test_encode_in_chunks(self, case, sizes): encoding, errors, text = case expected = _outcome(lambda: text.encode(encoding, errors)) @@ -305,7 +356,7 @@ def test_decoder_state_round_trip(self, case, data, split, big_endian): self.assertIsInstance(state[0], bytes) self.assertIsInstance(state[1], int) if not state[1]: - # Nothing but buffered bytes: feeding them to a fresh decoder + # Nothing but buffered bytes: after a reset, feeding them back # gives no output and reproduces the state. decoder.setstate((b"", 0)) self.assertEqual(decoder.decode(state[0]), "") @@ -368,6 +419,16 @@ def test_bom_codecs_reject_bare_input(self, encoding, data): decoder.decode(data, final=True) +# html.unescape resolves "&" sequences, reads references to the C1 range +# as cp1252 and drops references to the other control characters and to +# noncharacters, so none of those can be in text that is to come back +# from xmlcharrefreplace. +xml_safe_text = st.text( + alphabet=st.characters(exclude_categories=["Cc", "Cn", "Cs"], + exclude_characters="&"), + max_size=64) + + class RoundTripTests(unittest.TestCase): @hypothesis.given(case=round_trip_case) @@ -380,6 +441,29 @@ class RoundTripTests(unittest.TestCase): @hypothesis.example(case=("punycode", "b\xfccher")) @hypothesis.example(case=("idna", "b\xfccher.example")) @hypothesis.example(case=("unicode_escape", "\\n\n\u20ac")) + # gh-160302: the ISO-2022 encoders write U+001B as a bare ESC byte, + # which their decoders reject or read as the start of a sequence; + # iso2022_kr does the same with the shift characters U+000E and U+000F. + @hypothesis.example(case=("iso2022_jp", "\x1b")).xfail( + reason="gh-160302", raises=(AssertionError, UnicodeDecodeError)) + @hypothesis.example(case=("iso2022_jp_2004", "\x1b\xa0")).xfail( + reason="gh-160302", raises=(AssertionError, UnicodeDecodeError)) + @hypothesis.example(case=("iso2022_kr", "\x0e")).xfail( + reason="gh-160302", raises=(AssertionError, UnicodeDecodeError)) + @hypothesis.example(case=("iso2022_kr", "a\x0f")).xfail( + reason="gh-160302", raises=(AssertionError, UnicodeDecodeError)) + # gh-101863 (closed as not a bug): euc_kr encodes the Hangul filler + # U+3164 as the two bytes that start an eight-byte composed syllable, + # and will not decode them on their own. + @hypothesis.example(case=("euc_kr", "\u3164")).xfail( + reason="gh-101863", raises=UnicodeDecodeError) + @hypothesis.example(case=("euc_kr", "a\u3164b")).xfail( + reason="gh-101863", raises=UnicodeDecodeError) + # gh-160306: iso2022_jp_3 encodes with the JIS X 0213:2000 table but + # decodes with the 2004 one, where the same code point means U+9B1C, + # which the 2000 table then cannot encode again. + @hypothesis.example(case=("iso2022_jp_3", "\u9b1d")).xfail( + reason="gh-160306", raises=(AssertionError, UnicodeEncodeError)) def test_encode_decode(self, case): encoding, text = case try: @@ -432,13 +516,10 @@ def test_backslashreplace_decodes_back(self, case): self.assertEqual(as_ascii.decode("unicode_escape"), text) @hypothesis.given(case=handler_encoding.flatmap( - # html.unescape maps "&" sequences and treats the C1 controls as - # cp1252, so keep both out of the text. - lambda e: st.tuples(st.just(e), _round_trip_text(e, text.map( - lambda s: "".join(c for c in s - if c != "&" and not "\x80" <= c <= "\x9f")))))) + lambda e: st.tuples(st.just(e), _round_trip_text(e, xml_safe_text)))) @hypothesis.example(case=("ascii", "")) @hypothesis.example(case=("ascii", "a<\xe9>\u20ac\U0001f600")) + @hypothesis.example(case=("ascii", "\ufdcf\ufdf0\ufffd")) @hypothesis.example(case=("utf_7", "+\u20ac-")) @hypothesis.example(case=("shift_jis_2004", "\u3042\u20ac")) @hypothesis.example(case=("iso2022_kr", "\uac00\u20ac")) @@ -460,14 +541,15 @@ def test_escape_codec(self, data): @hypothesis.example(text="") @hypothesis.example(text="\\\n\x00\xe9\u20ac\ud800\U0001f600") def test_unicode_escape_with_surrogates(self, text): - for encoding in ("unicode_escape", "raw_unicode_escape"): - with self.subTest(encoding=encoding): - encoded = text.encode(encoding) - self.assertEqual(encoded.decode(encoding), - text.encode(encoding).decode(encoding)) + # Both escape codecs accept lone surrogates. unicode_escape writes + # ASCII and gives anything back; raw_unicode_escape only escapes + # what is not Latin-1, so it gives back text without backslashes. encoded = text.encode("unicode_escape") self.assertTrue(encoded.isascii()) self.assertEqual(encoded.decode("unicode_escape"), text) + raw_text = text.replace("\\", "") + encoded = raw_text.encode("raw_unicode_escape") + self.assertEqual(encoded.decode("raw_unicode_escape"), raw_text) @hypothesis.given(encoding=st.sampled_from(bytes_transform_encodings), data=data) @@ -489,10 +571,17 @@ def test_rot13(self, text): self.assertEqual(codecs.encode(encoded, "rot_13"), text) +# U+FFFE marks an undefined byte in a decoding table. UNDEFINED = "\ufffe" decoding_table = st.lists( st.characters() | st.just(UNDEFINED), min_size=256, max_size=256, ).map("".join) +# An injective table, so that encoding inverts decoding. charmap_build +# skips the undefined marker, so it cannot be an entry here. +encoding_table = st.lists( + st.characters(exclude_characters=UNDEFINED), min_size=256, max_size=256, + unique=True, +).map("".join) def _charmap_oracle(data, errors, table): @@ -541,18 +630,16 @@ def test_decode_matches_table(self, table, data, errors): _outcome(lambda: codecs.charmap_decode(data, errors, mapping)), _outcome(lambda: codecs.charmap_decode(data, errors, table))) - @hypothesis.given( - table=st.lists(st.characters(), min_size=256, max_size=256, - unique=True).map("".join), - indices=st.lists(st.integers(0, 255), max_size=32), - errors=st.sampled_from(ENCODE_ERRORS)) + @hypothesis.given(table=encoding_table, + indices=st.lists(st.integers(0, 255), max_size=32), + errors=st.sampled_from(ENCODE_ERRORS)) @hypothesis.example(table="".join(map(chr, range(256))), indices=[], errors="strict") @hypothesis.example(table="".join(map(chr, range(256, 512))), indices=[0, 255, 1], errors="strict") def test_encode_inverts_table(self, table, indices, errors): - # With an injective table, encoding characters from the table gives - # back their positions, through both kinds of encoding map. + # Encoding characters from the table gives back their positions, + # through both kinds of encoding map. text = "".join(table[i] for i in indices) expected = bytes(indices) built = codecs.charmap_build(table) @@ -564,11 +651,8 @@ def test_encode_inverts_table(self, table, indices, errors): self.assertEqual(codecs.charmap_decode(expected, "strict", table), (text, len(expected))) - @hypothesis.given( - table=st.lists(st.characters(), min_size=256, max_size=256, - unique=True).map("".join), - text=text_with_surrogates, - errors=st.sampled_from(ENCODE_ERRORS)) + @hypothesis.given(table=encoding_table, text=text_with_surrogates, + errors=st.sampled_from(ENCODE_ERRORS)) @hypothesis.example(table="".join(map(chr, range(256))), text="\u20ac", errors="strict") @hypothesis.example(table="".join(map(chr, range(256))), @@ -590,12 +674,21 @@ def test_encode_maps_agree(self, table, text, errors): alphabet=st.sampled_from("ab \n\r\x0b\x0c\x1c\x1d\x1e\x85\u2028\u2029"), max_size=64) +# Complete encoded text, with the error handler the reader will be given. +# Nothing else is generated: StreamReader never tells the decoder that the +# stream has ended, so an incomplete trailing sequence is dropped instead +# of reported (gh-160309, the same mechanism as gh-56717); the +# expected-failure examples on test_reader_matches_decode pin that. +reader_case = incremental_encoding.flatmap( + lambda e: st.tuples(st.just(e), st.sampled_from(DECODE_ERRORS), + _round_trip_text(e).map( + lambda s: s.encode(e, "replace")))) + class StreamTests(unittest.TestCase): @hypothesis.given(encoding=incremental_encoding, text=text, - write_sizes=sizes, - read_sizes=st.lists(st.integers(-1, 8), max_size=16)) + write_sizes=sizes, read_sizes=read_sizes) @hypothesis.example(encoding="utf_8", text="", write_sizes=[], read_sizes=[]) @hypothesis.example(encoding="utf_8", text="a\u20ac\U0001f600", @@ -606,8 +699,12 @@ class StreamTests(unittest.TestCase): read_sizes=[3]) @hypothesis.example(encoding="iso2022_jp", text="\u3042a\u3044", write_sizes=[1, 1], read_sizes=[1, 1, 1]) + @hypothesis.example(encoding="big5hkscs", text="\u00ca\u0304", + write_sizes=[], read_sizes=[1, 1]) def test_writer_then_reader(self, encoding, text, write_sizes, read_sizes): + # The size passed to read() is a number of bytes, so a read may + # well return more characters than that. try: expected = text.encode(encoding).decode(encoding) except UnicodeError: @@ -621,13 +718,78 @@ def test_writer_then_reader(self, encoding, text, write_sizes, stream.seek(0) reader = codecs.getreader(encoding)(stream) parts = [reader.read(n) for n in read_sizes] - for n, part in zip(read_sizes, parts): - if n >= 0: - self.assertLessEqual(len(part), n) parts.append(reader.read()) self.assertEqual("".join(parts), expected) self.assertEqual(reader.read(), "") + @hypothesis.given(case=reader_case, read_sizes=read_sizes) + @hypothesis.example(case=("utf_8", "strict", b""), read_sizes=[]) + @hypothesis.example(case=("utf_8", "replace", b"a\xe2\x82\xac"), + read_sizes=[1, 2]) + @hypothesis.example(case=("utf_16", "strict", b"\xff\xfea\x00"), + read_sizes=[1]) + @hypothesis.example(case=("euc_jp", "ignore", b"\xa4\xa2a"), + read_sizes=[1, 1]) + @hypothesis.example(case=("unicode_escape", "strict", b"a\\u00e9"), + read_sizes=[2]) + # gh-160309: a chopped trailing character is dropped rather than + # reported, whatever the error handler. + @hypothesis.example(case=("utf_8", "strict", b"ab\xe2"), + read_sizes=[]).xfail( + reason="gh-160309", raises=AssertionError) + @hypothesis.example(case=("utf_8", "replace", b"a\xe2\x82"), + read_sizes=[1]).xfail( + reason="gh-160309", raises=AssertionError) + @hypothesis.example(case=("utf_16", "strict", b"\xff\xfeB"), + read_sizes=[]).xfail( + reason="gh-160309", raises=AssertionError) + @hypothesis.example(case=("utf_32_be", "replace", b"\x00\x000"), + read_sizes=[2]).xfail( + reason="gh-160309", raises=AssertionError) + @hypothesis.example(case=("raw_unicode_escape", "strict", b"a\\u30"), + read_sizes=[]).xfail( + reason="gh-160309", raises=AssertionError) + def test_reader_matches_decode(self, case, read_sizes): + encoding, errors, data = case + expected = _outcome(lambda: data.decode(encoding, errors)) + + def read_all(): + reader = codecs.getreader(encoding)(io.BytesIO(data), errors) + parts = [reader.read(n) for n in read_sizes] + parts.append(reader.read()) + return "".join(parts) + + self.assertEqual(_outcome(read_all), expected) + + @hypothesis.given( + encoding=st.sampled_from(PYTHON_READER_ENCODINGS), text=text, + reads=st.lists(st.tuples(st.sampled_from([-1, 1, 2, 3, 8]), + st.integers(0, 8)), max_size=16)) + @hypothesis.example(encoding="utf_8", text="", reads=[]) + @hypothesis.example(encoding="utf_8", text="a\u20ac\U0001f600b", + reads=[(1, 2), (-1, 0), (8, 1), (1, 8)]) + @hypothesis.example(encoding="utf_16", text="a\u20acb", + reads=[(1, 1), (3, 2)]) + @hypothesis.example(encoding="utf_7", text="a\xe9\xe9b", + reads=[(1, 3), (2, 1)]) + @hypothesis.example(encoding="unicode_escape", text="\\a\u20ac", + reads=[(1, 1), (1, 1)]) + def test_reader_read_chars(self, encoding, text, reads): + # Asked for a number of characters, the Python StreamReader keeps + # reading until it has them, however many bytes it takes at a time + # (the multibyte codecs' readers only take a byte count). + try: + expected = text.encode(encoding).decode(encoding) + except UnicodeError: + return + reader = codecs.getreader(encoding)(io.BytesIO(text.encode(encoding))) + remaining = expected + for size, chars in reads: + part = reader.read(size, chars) + self.assertEqual(part, remaining[:chars]) + remaining = remaining[len(part):] + self.assertEqual(reader.read(), remaining) + @hypothesis.given(text=line_text, keepends=st.booleans()) @hypothesis.example(text="", keepends=True) @hypothesis.example(text="a\r\nb\rc\nd", keepends=True) @@ -716,140 +878,5 @@ def test_lookup_name_normalization(self, encoding, separators, upper): self.assertEqual("x".encode(variant), "x".encode(encoding)) -# The expected failures below each generate only inputs that are known to -# fail, so that they start passing as a whole once the bug is fixed. - -# Printable ASCII, which every codec below encodes to itself. -printable_ascii = st.characters(min_codepoint=0x20, max_codepoint=0x7e) -ascii_text = st.text(alphabet=printable_ascii, max_size=32) -ascii_bytes = st.binary(max_size=32).map(lambda b: bytes(c & 0x7f for c in b)) - - -def _incomplete_long_sequence(encoding): - # The first two bytes of a three- or four-byte sequence, where the - # second byte is ASCII: the stateless decoder escapes the first byte - # and decodes the second as itself. - if encoding == "gb18030": - return st.tuples(st.integers(0x81, 0xfe), st.integers(0x30, 0x39)) - return st.tuples(st.just(0x8f), st.integers(0, 0x7f)) - - -class MultibyteSurrogateescapeBug(unittest.TestCase): - """An incomplete sequence of three or four bytes at the end of the - input makes the incremental decoder drop whatever follows the first - escaped byte, where the stateless decoder escapes that byte and carries - on (gh-160307). Carved out of the chunked decoding tests above by - _decode_errors.""" - - @unittest.expectedFailure - @hypothesis.given( - case=st.sampled_from(sorted(LONG_SEQUENCE_ENCODINGS)).flatmap( - lambda e: st.tuples(st.just(e), ascii_bytes, - _incomplete_long_sequence(e)))) - @hypothesis.example(case=("gb18030", b"", (0x80, ord("6")))) - @hypothesis.example(case=("euc_jp", b"a", (0x8f, 0))) - @hypothesis.example(case=("euc_jis_2004", b"", (0x8f, ord("l")))) - def test_trailing_bytes_after_escape(self, case): - encoding, prefix, tail = case - data = prefix + bytes(tail) - expected = _outcome(lambda: data.decode(encoding, "surrogateescape")) - decoder = codecs.getincrementaldecoder(encoding)("surrogateescape") - self.assertEqual(_outcome(lambda: decoder.decode(data, True)), - expected) - - -class RoundTripBugs(unittest.TestCase): - """Characters an encoder accepts but its decoder cannot give back; see - ENCODE_ONLY_CHARACTERS for how the round-trip tests avoid them.""" - - def _check_round_trip(self, encoding, text): - encoded = text.encode(encoding) - self.assertEqual(encoded.decode(encoding), text) - - # The ISO-2022 encoders pass U+001B through as a bare ESC byte, which - # their own decoders then reject or misread as an escape sequence; - # iso2022_kr does the same with its shift characters U+000E and U+000F - # (gh-160302). The control character goes last, where it never forms - # a sequence. - @unittest.expectedFailure - @hypothesis.given( - case=st.sampled_from(sorted(ISO2022_ENCODINGS)).flatmap( - lambda e: st.tuples( - st.just(e), ascii_text, - st.sampled_from(ENCODE_ONLY_CHARACTERS[e])))) - @hypothesis.example(case=("iso2022_jp", "", "\x1b")) - @hypothesis.example(case=("iso2022_jp_2004", "a", "\x1b")) - @hypothesis.example(case=("iso2022_kr", "", "\x0e")) - @hypothesis.example(case=("iso2022_kr", "a", "\x0f")) - def test_iso2022_escape(self, case): - encoding, prefix, control = case - self._check_round_trip(encoding, prefix + control) - - # euc_kr encodes the Hangul filler U+3164 as the two bytes that start - # an eight-byte composed syllable, and will not decode them alone - # (gh-101863, closed as not a bug; recorded here as the asymmetry it is). - @unittest.expectedFailure - @hypothesis.given(prefix=ascii_text, suffix=ascii_text) - @hypothesis.example(prefix="", suffix="") - @hypothesis.example(prefix="a", suffix="b") - def test_euc_kr_filler(self, prefix, suffix): - self._check_round_trip("euc_kr", prefix + "\u3164" + suffix) - - # iso2022_jp_3 encodes with the JIS X 0213:2000 table but decodes with - # the 2004 one, where the same code point means U+9B1C (gh-160306). - @unittest.expectedFailure - @hypothesis.given(prefix=ascii_text, suffix=ascii_text) - @hypothesis.example(prefix="", suffix="") - @hypothesis.example(prefix="a", suffix="b") - def test_iso2022_jp_3_mapping(self, prefix, suffix): - self._check_round_trip("iso2022_jp_3", prefix + "\u9b1d" + suffix) - - -class StreamReaderTruncationBug(unittest.TestCase): - """StreamReader.read() never tells the decoder that the stream has - ended, so an incomplete trailing sequence is silently dropped instead - of raising (or being replaced), as it is everywhere else (gh-160309, - the same mechanism as gh-56717).""" - - @unittest.expectedFailure - @hypothesis.given( - encoding=st.sampled_from(["utf_8", "utf_16_le", "utf_32_be", - "utf_16", "raw_unicode_escape"]), - # No backslashes: raw_unicode_escape leaves them alone, and an even - # number in front of the chopped escape turns it into literal text. - text=text.map(lambda s: s.replace("\\", "")), - errors=st.sampled_from(["strict", "replace"])) - @hypothesis.example(encoding="utf_8", text="", errors="strict") - @hypothesis.example(encoding="utf_8", text="a", errors="replace") - @hypothesis.example(encoding="utf_16", text="\u3042", errors="strict") - def test_read_reports_truncated_input(self, encoding, text, errors): - # Chop the last byte off a character that takes more than one. - encoded = (text + "\u3042").encode(encoding)[:-1] - expected = _outcome(lambda: encoded.decode(encoding, errors)) - reader = codecs.getreader(encoding)(io.BytesIO(encoded), errors) - self.assertEqual(_outcome(lambda: reader.read() + reader.read()), - expected) - - -class PunycodeIncrementalBug(unittest.TestCase): - """The punycode incremental codecs encode and decode each chunk on its - own, so anything fed in more than one piece comes out wrong - (gh-160308). Punycode is left out of the chunked tests above as a - whole-string codec.""" - - # Two non-empty ASCII chunks give two "-" terminated labels where the - # whole string gives one. - @unittest.expectedFailure - @hypothesis.given( - head=st.text(alphabet=printable_ascii, min_size=1, max_size=32), - tail=st.text(alphabet=printable_ascii, min_size=1, max_size=32)) - @hypothesis.example(head="a", tail="b") - @hypothesis.example(head="b\xfc", tail="cher") - def test_chunked_encode(self, head, tail): - encoder = codecs.getincrementalencoder("punycode")() - chunked = encoder.encode(head) + encoder.encode(tail, final=True) - self.assertEqual(chunked, (head + tail).encode("punycode")) - - if __name__ == "__main__": unittest.main()