gh-156965: Make the utf-7-imap incremental decoder and stream reader stateful

Lib/encodings/utf_7_imap.py

@@ -54,7 +54,7 @@ def flush(end):
     flush(len(input))
     return res.take_bytes(), len(input)
 
-def utf_7_imap_decode(input, errors='strict'):
+def utf_7_imap_decode(input, errors='strict', final=True):
     if errors != 'strict':
         raise UnicodeError(f"Unsupported error handling: {errors}")
     input = bytes(input)
@@ -71,6 +71,8 @@ def flush(end):
             flush(i)
             j = input.find(b'-', i + 1)
             if j < 0:
+                if not final:
+                    return ''.join(res), i
                 raise UnicodeDecodeError('utf-7-imap', input, i, n,
                                          'unterminated shift sequence')
             if j == i + 1:          # '&-'
@@ -106,15 +108,16 @@ class IncrementalEncoder(codecs.IncrementalEncoder):
     def encode(self, input, final=False):
         return utf_7_imap_encode(input, self.errors)[0]
 
-class IncrementalDecoder(codecs.IncrementalDecoder):
-    def decode(self, input, final=False):
-        return utf_7_imap_decode(input, self.errors)[0]
+class IncrementalDecoder(codecs.BufferedIncrementalDecoder):
+    def _buffer_decode(self, input, errors, final):
+        return utf_7_imap_decode(input, errors, final)
 
 class StreamWriter(Codec, codecs.StreamWriter):
     pass
 
 class StreamReader(Codec, codecs.StreamReader):
-    pass
+    def decode(self, input, errors='strict'):
+        return utf_7_imap_decode(input, errors, False)
 
 ### encodings module API
 

Lib/test/test_codecs.py

@@ -1467,6 +1467,24 @@ def test_only_strict_errors(self):
         with self.assertRaises(UnicodeError):
             b'x'.decode('utf-7-imap', 'ignore')
 
+    @support.subTests('uni,encoded', utf_7_imap_testcases)
+    def test_incremental_decode(self, uni, encoded):
+        # A shift sequence split across chunks is not an error: the joined
+        # output has to match the stateless decoder.
+        decoder = codecs.getincrementaldecoder('utf-7-imap')()
+        out = ''.join(decoder.decode(encoded[i:i + 1])
+                      for i in range(len(encoded)))
+        self.assertEqual(out + decoder.decode(b'', True), uni)
+
+    def test_stream_read(self):
+        # Reading a fixed number of characters stops inside a shift sequence.
+        uni = '\u53f0' * 2000
+        encoded = uni.encode('utf-7-imap')
+        with io.TextIOWrapper(io.BytesIO(encoded), encoding='utf-7-imap') as f:
+            self.assertEqual(f.read(100), uni[:100])
+        chunks = [encoded[i:i + 16] for i in range(0, len(encoded), 16)]
+        self.assertEqual(''.join(codecs.iterdecode(chunks, 'utf-7-imap')), uni)
+
     def test_stateless(self):
         # The codec is registered and exposes the standard interface.
         info = codecs.lookup('utf-7-imap')