Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 7 additions & 9 deletions Lib/encodings/utf_8_sig.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,12 +16,10 @@ def encode(input, errors='strict'):
len(input))

def decode(input, errors='strict'):
prefix = 0
if input[:3] == codecs.BOM_UTF8:
input = input[3:]
prefix = 3
(output, consumed) = codecs.utf_8_decode(input, errors, True)
return (output, consumed+prefix)
(output, consumed) = codecs.utf_8_decode(input, errors, True)
return (output[1:], consumed)
return codecs.utf_8_decode(input, errors, True)

class IncrementalEncoder(codecs.IncrementalEncoder):
def __init__(self, errors='strict'):
Expand Down Expand Up @@ -64,8 +62,8 @@ def _buffer_decode(self, input, errors, final):
self.first = 0
if input[:3] == codecs.BOM_UTF8:
(output, consumed) = \
codecs.utf_8_decode(input[3:], errors, final)
return (output, consumed+3)
codecs.utf_8_decode(input, errors, final)
return (output[1:], consumed)
return codecs.utf_8_decode(input, errors, final)

def reset(self):
Expand Down Expand Up @@ -110,8 +108,8 @@ def decode(self, input, errors='strict'):
return ("", 0)
elif input[:3] == codecs.BOM_UTF8:
self.decode = codecs.utf_8_decode
(output, consumed) = codecs.utf_8_decode(input[3:],errors)
return (output, consumed+3)
(output, consumed) = codecs.utf_8_decode(input, errors)
return (output[1:], consumed)
# (else) no BOM present
self.decode = codecs.utf_8_decode
return codecs.utf_8_decode(input, errors)
Expand Down
53 changes: 53 additions & 0 deletions Lib/test/test_codecs.py
Original file line number Diff line number Diff line change
Expand Up @@ -1169,6 +1169,59 @@ def test_stream_bare(self):
got = ostream.getvalue()
self.assertEqual(got, unistring)

def test_decode_error_offsets_with_bom(self):
# gh-159071: UnicodeDecodeError offsets should count the BOM
data = codecs.BOM_UTF8 + b"abc\xff"
with self.assertRaises(UnicodeDecodeError) as cm:
data.decode("utf-8-sig")
exc = cm.exception
self.assertEqual(exc.object, data)
self.assertEqual(exc.start, 6)
self.assertEqual(exc.end, 7)

# Test error right after BOM
data = codecs.BOM_UTF8 + b"\xff"
with self.assertRaises(UnicodeDecodeError) as cm:
data.decode("utf-8-sig")
exc = cm.exception
self.assertEqual(exc.object, data)
self.assertEqual(exc.start, 3)
self.assertEqual(exc.end, 4)

# Test IncrementalDecoder
dec = codecs.getincrementaldecoder("utf-8-sig")()
with self.assertRaises(UnicodeDecodeError) as cm:
dec.decode(data, final=True)
exc = cm.exception
self.assertEqual(exc.object, data)
self.assertEqual(exc.start, 3)
self.assertEqual(exc.end, 4)

# Test StreamReader
reader = codecs.getreader("utf-8-sig")(io.BytesIO(data))
with self.assertRaises(UnicodeDecodeError) as cm:
reader.read()
exc = cm.exception
self.assertEqual(exc.object, data)
self.assertEqual(exc.start, 3)
self.assertEqual(exc.end, 4)

# Test custom error handler receives correct offsets
received = []
def handler(err):
received.append((err.object, err.start, err.end))
return ("?", err.end)

old_handler = codecs.lookup_error("replace")
codecs.register_error("test_utf8sig_handler", handler)
try:
result = data.decode("utf-8-sig", "test_utf8sig_handler")
self.assertEqual(result, "?")
self.assertEqual(received, [(data, 3, 4)])
finally:
# clean up error handler if needed
pass


class EscapeDecodeTest(unittest.TestCase):
def test_empty(self):
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
Fixed :exc:`UnicodeDecodeError` offsets in the :mod:`encodings.utf_8_sig`
codec when input starts with a BOM.
Loading