https://github.com/python/cpython/commit/7e6b0c9f92151a0a3cf13cf2211a13ca8836a182
commit: 7e6b0c9f92151a0a3cf13cf2211a13ca8836a182
branch: 3.14
author: Miss Islington (bot) <[email protected]>
committer: gpshead <[email protected]>
date: 2026-10-31T13:24:30Z
summary:

[3.14] gh-160522: Restore CJK encoder state after a failed incremental encode 
(GH-160523) (#161001)

gh-160522: Restore CJK encoder state after a failed incremental encode 
(GH-160523)

A failed IncrementalEncoder.encode() or StreamWriter.write() of a
stateful CJK codec kept the shift state reached before the error
although its output was discarded.
(cherry picked from commit 5b2558292978113211c5a720873183b38547f3d0)

Co-authored-by: Gregory P. Smith <[email protected]>

files:
A Misc/NEWS.d/next/Library/2026-10-22-08-18-31.gh-issue-160522.cJkSt8.rst
M Lib/test/test_multibytecodec.py
M Modules/cjkcodecs/multibytecodec.c

diff --git a/Lib/test/test_multibytecodec.py b/Lib/test/test_multibytecodec.py
index 0b34c6e69fab6b6..8d5eb78608afc71 100644
--- a/Lib/test/test_multibytecodec.py
+++ b/Lib/test/test_multibytecodec.py
@@ -438,6 +438,48 @@ def test_incrementalencoder_final(self):
         self.assertEqual(output, self.expected_reset)
         self.assertEqual(encoder.encode('', final=True), b'')
 
+    def test_incrementalencoder_error(self):
+        # The output of a failed encode() is lost, so the shift state
+        # must not be changed.
+        encoder = codecs.getincrementalencoder(self.encoding)()
+        state = encoder.getstate()
+        for final in False, True:
+            self.assertRaises(UnicodeEncodeError, encoder.encode,
+                              self.text + '\U0001f600', final)
+            self.assertEqual(encoder.getstate(), state)
+        self.assertEqual(encoder.encode(self.text), self.expected)
+        state = encoder.getstate()
+        for final in False, True:
+            self.assertRaises(UnicodeEncodeError, encoder.encode,
+                              'a\U0001f600', final)
+            self.assertEqual(encoder.getstate(), state)
+        self.assertEqual(encoder.encode('a'), self.reset + b'a')
+
+    def test_streamwriter_error(self):
+        stream = io.BytesIO()
+        writer = codecs.getwriter(self.encoding)(stream)
+        self.assertRaises(UnicodeEncodeError, writer.write,
+                          self.text + '\U0001f600')
+        self.assertEqual(stream.getvalue(), b'')
+        writer.write(self.text)
+        self.assertEqual(stream.getvalue(), self.expected)
+
+    def test_textiowrapper_error(self):
+        raw = io.BytesIO()
+        with io.TextIOWrapper(raw, encoding=self.encoding) as f:
+            self.assertRaises(UnicodeEncodeError, f.write,
+                              self.text + '\U0001f600')
+            f.write(self.text)
+            f.flush()
+            self.assertEqual(raw.getvalue(), self.expected)
+
+class TestISO2022KRStateful(TestStateful):
+    text = '\uac00\uac00'
+    encoding = 'iso-2022-kr'
+    expected = b'\x1b$)C\x0e0!0!'
+    reset = b'\x0f'
+    expected_reset = expected + reset
+
 class TestHZStateful(TestStateful):
     text = '\u804a\u804a'
     encoding = 'hz'
diff --git 
a/Misc/NEWS.d/next/Library/2026-10-22-08-18-31.gh-issue-160522.cJkSt8.rst 
b/Misc/NEWS.d/next/Library/2026-10-22-08-18-31.gh-issue-160522.cJkSt8.rst
new file mode 100644
index 000000000000000..ae93269f8714760
--- /dev/null
+++ b/Misc/NEWS.d/next/Library/2026-10-22-08-18-31.gh-issue-160522.cJkSt8.rst
@@ -0,0 +1,4 @@
+Fix the incremental encoders and stream writers of the stateful CJK codecs
+(``iso2022_*`` and ``hz``) keeping a changed shift state after an
+:meth:`!encode` or :meth:`!write` call failed.  The next call could omit the
+escape sequence, writing multibyte characters in ASCII mode.
diff --git a/Modules/cjkcodecs/multibytecodec.c 
b/Modules/cjkcodecs/multibytecodec.c
index 1841f62c7797c20..e106e4d546532eb 100644
--- a/Modules/cjkcodecs/multibytecodec.c
+++ b/Modules/cjkcodecs/multibytecodec.c
@@ -801,6 +801,7 @@ encoder_encode_stateful(MultibyteStatefulEncoderContext 
*ctx,
     PyObject *inbuf = NULL;
     Py_ssize_t inpos, datalen;
     PyObject *origpending = NULL;
+    MultibyteCodec_State origstate;
 
     if (PyUnicode_Check(unistr))
         ucvt = NULL;
@@ -816,6 +817,7 @@ encoder_encode_stateful(MultibyteStatefulEncoderContext 
*ctx,
         }
     }
 
+    origstate = ctx->state;
     if (ctx->pending) {
         PyObject *inbuf_tmp;
 
@@ -873,6 +875,9 @@ encoder_encode_stateful(MultibyteStatefulEncoderContext 
*ctx,
     return r;
 
 errorexit:
+    /* recover the original codec state: the discarded output may have
+     * included an escape sequence for a state change */
+    ctx->state = origstate;
     Py_XDECREF(r);
     Py_XDECREF(ucvt);
     Py_XDECREF(origpending);

_______________________________________________
Python-checkins mailing list -- [email protected]
To unsubscribe send an email to [email protected]
https://mail.python.org/mailman3//lists/python-checkins.python.org
Member address: [email protected]

Reply via email to