from ctypes import POINTER, WINFUNCTYPE, WinDLL

from ctypes.wintypes import BOOL, BYTE, WCHAR, UINT, DWORD

MAX_LEADBYTES = 12 # 5 ranges, 2 bytes ea., 0 term.

MAX_DEFAULTCHAR = 2 # single or double byte

MAX_PATH = 260

class CPINFOEXW(ctypes.Structure):

_fields_ = [("MaxCharSize", UINT),

("DefaultChar", BYTE*MAX_DEFAULTCHAR),

("LeadByte", BYTE*MAX_LEADBYTES),

("UnicodeDefaultChar", WCHAR),

("CodePage", UINT),

("CodePageName", WCHAR*MAX_PATH)]

prototype = WINFUNCTYPE(BOOL, UINT, DWORD, POINTER(CPINFOEXW))

GetCPInfoEx = prototype(("GetCPInfoExW", WinDLL("kernel32")))

info = CPINFOEXW()

"""

def __init__(self, buffer):

self._buffer = buffer

def write(self, chars):

self._buffer += chars

def read(self, size=-1):

if size<0:

s = self._buffer

self._buffer = self._buffer[:0] # make empty

return s

else:

s = self._buffer[:size]

self._buffer = self._buffer[size:]

def check_state_handling_decode(self, encoding, u, s):

for i in range(len(s)+1):

d = codecs.getincrementaldecoder(encoding)()

part1 = d.decode(s[:i])

state = d.getstate()

self.assertIsInstance(state[1], int)

# Check that the condition stated in the documentation for

# IncrementalDecoder.getstate() holds

if not state[1]:

# reset decoder to the default state without anything buffered

d.setstate((state[0][:0], 0))

# Feeding the previous input may not produce any output

self.assertTrue(not d.decode(state[0]))

# The decoder must return to the same state

self.assertEqual(state, d.getstate())

# Create a new decoder and set it to the state

# we extracted from the old one

d = codecs.getincrementaldecoder(encoding)()

d.setstate(state)

part2 = d.decode(s[i:], True)

self.assertEqual(u, part1+part2)

def check_state_handling_encode(self, encoding, u, s):

for i in range(len(u)+1):

d = codecs.getincrementalencoder(encoding)()

part1 = d.encode(u[:i])

state = d.getstate()

d = codecs.getincrementalencoder(encoding)()

d.setstate(state)

part2 = d.encode(u[i:], True)

def check_partial(self, input, partialresults):

# get a StreamReader for the encoding and feed the bytestring version

# of input to the reader byte by byte. Read everything available from

# the StreamReader and check that the results equal the appropriate

# entries from partialresults.

q = Queue(b"")

r = codecs.getreader(self.encoding)(q)

result = ""

for (c, partialresult) in zip(input.encode(self.encoding), partialresults, strict=True):

q.write(bytes([c]))

result += r.read()

self.assertEqual(result, partialresult)

# check that there's nothing left in the buffers

self.assertEqual(r.read(), "")

self.assertEqual(r.bytebuffer, b"")

# do the check again, this time using an incremental decoder

d = codecs.getincrementaldecoder(self.encoding)()

result = ""

for (c, partialresult) in zip(input.encode(self.encoding), partialresults, strict=True):

result += d.decode(bytes([c]))

self.assertEqual(result, partialresult)

# check that there's nothing left in the buffers

self.assertEqual(d.decode(b"", True), "")

self.assertEqual(d.buffer, b"")

# Check whether the reset method works properly

d.reset()

result = ""

for (c, partialresult) in zip(input.encode(self.encoding), partialresults, strict=True):

result += d.decode(bytes([c]))

self.assertEqual(result, partialresult)

# check that there's nothing left in the buffers

self.assertEqual(d.decode(b"", True), "")

self.assertEqual(d.buffer, b"")

# check iterdecode()

encoded = input.encode(self.encoding)

self.assertEqual(

input,

"".join(codecs.iterdecode([bytes([c]) for c in encoded], self.encoding))

)

def test_readline(self):

def getreader(input):

stream = io.BytesIO(input.encode(self.encoding))

return codecs.getreader(self.encoding)(stream)

def readalllines(input, keepends=True, size=None):

reader = getreader(input)

lines = []

while True:

line = reader.readline(size=size, keepends=keepends)

if not line:

break

lines.append(line)

return "|".join(lines)

s = "foo\nbar\r\nbaz\rspam\u2028eggs"

sexpected = "foo\n|bar\r\n|baz\r|spam\u2028|eggs"

sexpectednoends = "foo|bar|baz|spam|eggs"

self.assertEqual(readalllines(s, True), sexpected)

self.assertEqual(readalllines(s, False), sexpectednoends)

self.assertEqual(readalllines(s, True, 10), sexpected)

self.assertEqual(readalllines(s, False, 10), sexpectednoends)

lineends = ("\n", "\r\n", "\r", "\u2028")

# Test long lines (multiple calls to read() in readline())

vw = []

vwo = []

for (i, lineend) in enumerate(lineends):

vw.append((i*200+200)*"\u3042" + lineend)

vwo.append((i*200+200)*"\u3042")

self.assertEqual(readalllines("".join(vw), True), "|".join(vw))

self.assertEqual(readalllines("".join(vw), False), "|".join(vwo))

# Test lines where the first read might end with \r, so the

# reader has to look ahead whether this is a lone \r or a \r\n

for size in range(80):

for lineend in lineends:

s = 10*(size*"a" + lineend + "xxx\n")

reader = getreader(s)

for i in range(10):

self.assertEqual(

reader.readline(keepends=True),

size*"a" + lineend,

)

self.assertEqual(

reader.readline(keepends=True),

"xxx\n",

)

reader = getreader(s)

for i in range(10):

self.assertEqual(

reader.readline(keepends=False),

size*"a",

)

self.assertEqual(

reader.readline(keepends=False),

"xxx",

)

def test_mixed_readline_and_read(self):

lines = ["Humpty Dumpty sat on a wall,\n",

"Humpty Dumpty had a great fall.\r\n",

"All the king's horses and all the king's men\r",

"Couldn't put Humpty together again."]

data = ''.join(lines)

def getreader():

stream = io.BytesIO(data.encode(self.encoding))

return codecs.getreader(self.encoding)(stream)

# Issue #8260: Test readline() followed by read()

f = getreader()

self.assertEqual(f.readline(), lines[0])

self.assertEqual(f.read(), ''.join(lines[1:]))

self.assertEqual(f.read(), '')

# Issue #32110: Test readline() followed by read(n)

f = getreader()

self.assertEqual(f.readline(), lines[0])

self.assertEqual(f.read(1), lines[1][0])

self.assertEqual(f.read(0), '')

self.assertEqual(f.read(100), data[len(lines[0]) + 1:][:100])

# Issue #16636: Test readline() followed by readlines()

f = getreader()

self.assertEqual(f.readline(), lines[0])

self.assertEqual(f.readlines(), lines[1:])

self.assertEqual(f.read(), '')

# Test read(n) followed by read()

f = getreader()

self.assertEqual(f.read(size=40, chars=5), data[:5])

self.assertEqual(f.read(), data[5:])

self.assertEqual(f.read(), '')

# Issue #32110: Test read(n) followed by read(n)

f = getreader()

self.assertEqual(f.read(size=40, chars=5), data[:5])

self.assertEqual(f.read(1), data[5])

self.assertEqual(f.read(0), '')

self.assertEqual(f.read(100), data[6:106])

# Issue #12446: Test read(n) followed by readlines()

f = getreader()

self.assertEqual(f.read(size=40, chars=5), data[:5])

self.assertEqual(f.readlines(), [lines[0][5:]] + lines[1:])

self.assertEqual(f.read(), '')

def test_bug1175396(self):

s = [

'<%!--===================================================\r\n',

' BLOG index page: show recent articles,\r\n',

' today\'s articles, or articles of a specific date.\r\n',

'========================================================--%>\r\n',

'<%@inputencoding="ISO-8859-1"%>\r\n',

'<%@pagetemplate=TEMPLATE.y%>\r\n',

'<%@import=import frog.util, frog%>\r\n',

'<%@import=import frog.objects%>\r\n',

'<%@import=from frog.storageerrors import StorageError%>\r\n',

'<%\r\n',

'\r\n',

'import logging\r\n',

'log=logging.getLogger("Snakelets.logger")\r\n',

'\r\n',

'\r\n',

'user=self.SessionCtx.user\r\n',

'storageEngine=self.SessionCtx.storageEngine\r\n',

'\r\n',

'\r\n',

'def readArticlesFromDate(date, count=None):\r\n',

' entryids=storageEngine.listBlogEntries(date)\r\n',

' entryids.reverse() # descending\r\n',

' if count:\r\n',

' entryids=entryids[:count]\r\n',

' try:\r\n',

' return [ frog.objects.BlogEntry.load(storageEngine, date, Id) for Id in entryids ]\r\n',

' except StorageError,x:\r\n',

' log.error("Error loading articles: "+str(x))\r\n',

' self.abort("cannot load articles")\r\n',

'\r\n',

'showdate=None\r\n',

'\r\n',

'arg=self.Request.getArg()\r\n',

'if arg=="today":\r\n',

' #-------------------- TODAY\'S ARTICLES\r\n',

' self.write("<h2>Today\'s articles</h2>")\r\n',

' showdate = frog.util.isodatestr() \r\n',

' entries = readArticlesFromDate(showdate)\r\n',

'elif arg=="active":\r\n',

' #-------------------- ACTIVE ARTICLES redirect\r\n',

' self.Yredirect("active.y")\r\n',

'elif arg=="login":\r\n',

' #-------------------- LOGIN PAGE redirect\r\n',

' self.Yredirect("login.y")\r\n',

'elif arg=="date":\r\n',

' #-------------------- ARTICLES OF A SPECIFIC DATE\r\n',

' showdate = self.Request.getParameter("date")\r\n',

' self.write("<h2>Articles written on %s</h2>"% frog.util.mediumdatestr(showdate))\r\n',

' entries = readArticlesFromDate(showdate)\r\n',

'else:\r\n',

' #-------------------- RECENT ARTICLES\r\n',

' self.write("<h2>Recent articles</h2>")\r\n',

' dates=storageEngine.listBlogEntryDates()\r\n',

' if dates:\r\n',

' entries=[]\r\n',

' SHOWAMOUNT=10\r\n',

' for showdate in dates:\r\n',

' entries.extend( readArticlesFromDate(showdate, SHOWAMOUNT-len(entries)) )\r\n',

' if len(entries)>=SHOWAMOUNT:\r\n',

' break\r\n',

' \r\n',

]

stream = io.BytesIO("".join(s).encode(self.encoding))

reader = codecs.getreader(self.encoding)(stream)

for (i, line) in enumerate(reader):

self.assertEqual(line, s[i])

def test_readlinequeue(self):

q = Queue(b"")

writer = codecs.getwriter(self.encoding)(q)

reader = codecs.getreader(self.encoding)(q)

# No lineends

writer.write("foo\r")

self.assertEqual(reader.readline(keepends=False), "foo")

writer.write("\nbar\r")

self.assertEqual(reader.readline(keepends=False), "")

self.assertEqual(reader.readline(keepends=False), "bar")

writer.write("baz")

self.assertEqual(reader.readline(keepends=False), "baz")

self.assertEqual(reader.readline(keepends=False), "")

# Lineends

writer.write("foo\r")

self.assertEqual(reader.readline(keepends=True), "foo\r")

writer.write("\nbar\r")

self.assertEqual(reader.readline(keepends=True), "\n")

self.assertEqual(reader.readline(keepends=True), "bar\r")

writer.write("baz")

self.assertEqual(reader.readline(keepends=True), "baz")

self.assertEqual(reader.readline(keepends=True), "")

writer.write("foo\r\n")

self.assertEqual(reader.readline(keepends=True), "foo\r\n")

def test_bug1098990_a(self):

s1 = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx yyyyyyyyyyyyyyyyyyyyyyyyyyyyyyy\r\n"

s2 = "offending line: ladfj askldfj klasdj fskla dfzaskdj fasklfj laskd fjasklfzzzzaa%whereisthis!!!\r\n"

s3 = "next line.\r\n"

s = (s1+s2+s3).encode(self.encoding)

stream = io.BytesIO(s)

reader = codecs.getreader(self.encoding)(stream)

self.assertEqual(reader.readline(), s1)

self.assertEqual(reader.readline(), s2)

self.assertEqual(reader.readline(), s3)

self.assertEqual(reader.readline(), "")

def test_bug1098990_b(self):

s1 = "aaaaaaaaaaaaaaaaaaaaaaaa\r\n"

s2 = "bbbbbbbbbbbbbbbbbbbbbbbb\r\n"

s3 = "stillokay:bbbbxx\r\n"

s4 = "broken!!!!badbad\r\n"

s5 = "againokay.\r\n"

s = (s1+s2+s3+s4+s5).encode(self.encoding)

stream = io.BytesIO(s)

reader = codecs.getreader(self.encoding)(stream)

self.assertEqual(reader.readline(), s1)

self.assertEqual(reader.readline(), s2)

self.assertEqual(reader.readline(), s3)

self.assertEqual(reader.readline(), s4)

self.assertEqual(reader.readline(), s5)

self.assertEqual(reader.readline(), "")

ill_formed_sequence_replace = "\ufffd"

def test_lone_surrogates(self):

self.assertRaises(UnicodeEncodeError, "\ud800".encode, self.encoding)

self.assertEqual("[\uDC80]".encode(self.encoding, "backslashreplace"),

"[\\udc80]".encode(self.encoding))

self.assertEqual("[\uDC80]".encode(self.encoding, "namereplace"),

"[\\udc80]".encode(self.encoding))

self.assertEqual("[\uDC80]".encode(self.encoding, "xmlcharrefreplace"),

"[&#56448;]".encode(self.encoding))

self.assertEqual("[\uDC80]".encode(self.encoding, "ignore"),

"[]".encode(self.encoding))

self.assertEqual("[\uDC80]".encode(self.encoding, "replace"),

"[?]".encode(self.encoding))

# sequential surrogate characters

self.assertEqual("[\uD800\uDC80]".encode(self.encoding, "ignore"),

"[]".encode(self.encoding))

self.assertEqual("[\uD800\uDC80]".encode(self.encoding, "replace"),

"[??]".encode(self.encoding))

bom = "".encode(self.encoding)

for before, after in [("\U00010fff", "A"), ("[", "]"),

("A", "\U00010fff")]:

before_sequence = before.encode(self.encoding)[len(bom):]

after_sequence = after.encode(self.encoding)[len(bom):]

test_string = before + "\uDC80" + after

test_sequence = (bom + before_sequence +

self.ill_formed_sequence + after_sequence)

self.assertRaises(UnicodeDecodeError, test_sequence.decode,

self.encoding)

self.assertEqual(test_string.encode(self.encoding,

"surrogatepass"),

test_sequence)

self.assertEqual(test_sequence.decode(self.encoding,

"surrogatepass"),

test_string)

self.assertEqual(test_sequence.decode(self.encoding, "ignore"),

before + after)

self.assertEqual(test_sequence.decode(self.encoding, "replace"),

before + self.ill_formed_sequence_replace + after)

backslashreplace = ''.join('\\x%02x' % b

for b in self.ill_formed_sequence)

self.assertEqual(test_sequence.decode(self.encoding, "backslashreplace"),

before + backslashreplace + after)

def test_incremental_surrogatepass(self):

# Test incremental decoder for surrogatepass handler:

# see issue #24214

# High surrogate

data = '\uD901'.encode(self.encoding, 'surrogatepass')

for i in range(1, len(data)):

dec = codecs.getincrementaldecoder(self.encoding)('surrogatepass')

self.assertEqual(dec.decode(data[:i]), '')

self.assertEqual(dec.decode(data[i:], True), '\uD901')

# Low surrogate

data = '\uDC02'.encode(self.encoding, 'surrogatepass')

for i in range(1, len(data)):

dec = codecs.getincrementaldecoder(self.encoding)('surrogatepass')

self.assertEqual(dec.decode(data[:i]), '')

encoding = "utf-32"

if sys.byteorder == 'little':

ill_formed_sequence = b"\x80\xdc\x00\x00"

else:

ill_formed_sequence = b"\x00\x00\xdc\x80"

spamle = (b'\xff\xfe\x00\x00'

b's\x00\x00\x00p\x00\x00\x00a\x00\x00\x00m\x00\x00\x00'

b's\x00\x00\x00p\x00\x00\x00a\x00\x00\x00m\x00\x00\x00')

spambe = (b'\x00\x00\xfe\xff'

b'\x00\x00\x00s\x00\x00\x00p\x00\x00\x00a\x00\x00\x00m'

b'\x00\x00\x00s\x00\x00\x00p\x00\x00\x00a\x00\x00\x00m')

def test_only_one_bom(self):

_,_,reader,writer = codecs.lookup(self.encoding)

# encode some stream

s = io.BytesIO()

f = writer(s)

f.write("spam")

f.write("spam")

d = s.getvalue()

# check whether there is exactly one BOM in it

self.assertTrue(d == self.spamle or d == self.spambe)

# try to read it back

s = io.BytesIO(d)

f = reader(s)

self.assertEqual(f.read(), "spamspam")

def test_badbom(self):

s = io.BytesIO(4*b"\xff")

f = codecs.getreader(self.encoding)(s)

self.assertRaises(UnicodeError, f.read)

s = io.BytesIO(8*b"\xff")

f = codecs.getreader(self.encoding)(s)

self.assertRaises(UnicodeError, f.read)

def test_partial(self):

self.check_partial(

"\x00\xff\u0100\uffff\U00010000",

[

"", # first byte of BOM read

"", # second byte of BOM read

"", # third byte of BOM read

"", # fourth byte of BOM read => byteorder known

"",

"",

"",

"\x00",

"\x00",

"\x00",

"\x00",

"\x00\xff",

"\x00\xff",

"\x00\xff",

"\x00\xff",

"\x00\xff\u0100",

"\x00\xff\u0100",

"\x00\xff\u0100",

"\x00\xff\u0100",

"\x00\xff\u0100\uffff",

"\x00\xff\u0100\uffff",

"\x00\xff\u0100\uffff",

"\x00\xff\u0100\uffff",

"\x00\xff\u0100\uffff\U00010000",

]

)

def test_handlers(self):

self.assertEqual(('\ufffd', 1),

codecs.utf_32_decode(b'\x01', 'replace', True))

self.assertEqual(('', 1),

codecs.utf_32_decode(b'\x01', 'ignore', True))

def test_errors(self):

self.assertRaises(UnicodeDecodeError, codecs.utf_32_decode,

b"\xff", "strict", True)

def test_decoder_state(self):

self.check_state_handling_decode(self.encoding,

"spamspam", self.spamle)

self.check_state_handling_decode(self.encoding,

"spamspam", self.spambe)

def test_issue8941(self):

# Issue #8941: insufficient result allocation when decoding into

# surrogate pairs on UCS-2 builds.

encoded_le = b'\xff\xfe\x00\x00' + b'\x00\x00\x01\x00' * 1024

self.assertEqual('\U00010000' * 1024,

codecs.utf_32_decode(encoded_le)[0])

encoded_be = b'\x00\x00\xfe\xff' + b'\x00\x01\x00\x00' * 1024

self.assertEqual('\U00010000' * 1024,

encoding = "utf-32-le"

ill_formed_sequence = b"\x80\xdc\x00\x00"

def test_partial(self):

self.check_partial(

"\x00\xff\u0100\uffff\U00010000",

[

"",

"",

"",

"\x00",

"\x00",

"\x00",

"\x00",

"\x00\xff",

"\x00\xff",

"\x00\xff",

"\x00\xff",

"\x00\xff\u0100",

"\x00\xff\u0100",

"\x00\xff\u0100",

"\x00\xff\u0100",

"\x00\xff\u0100\uffff",

"\x00\xff\u0100\uffff",

"\x00\xff\u0100\uffff",

"\x00\xff\u0100\uffff",

"\x00\xff\u0100\uffff\U00010000",

]

)

def test_simple(self):

self.assertEqual("\U00010203".encode(self.encoding), b"\x03\x02\x01\x00")

def test_errors(self):

self.assertRaises(UnicodeDecodeError, codecs.utf_32_le_decode,

b"\xff", "strict", True)

def test_issue8941(self):

# Issue #8941: insufficient result allocation when decoding into

# surrogate pairs on UCS-2 builds.

encoded = b'\x00\x00\x01\x00' * 1024

self.assertEqual('\U00010000' * 1024,

encoding = "utf-32-be"

ill_formed_sequence = b"\x00\x00\xdc\x80"

def test_partial(self):

self.check_partial(

"\x00\xff\u0100\uffff\U00010000",

[

"",

"",

"",

"\x00",

"\x00",

"\x00",

"\x00",

"\x00\xff",

"\x00\xff",

"\x00\xff",

"\x00\xff",

"\x00\xff\u0100",

"\x00\xff\u0100",

"\x00\xff\u0100",

"\x00\xff\u0100",

"\x00\xff\u0100\uffff",

"\x00\xff\u0100\uffff",

"\x00\xff\u0100\uffff",

"\x00\xff\u0100\uffff",

"\x00\xff\u0100\uffff\U00010000",

]

)

def test_simple(self):

self.assertEqual("\U00010203".encode(self.encoding), b"\x00\x01\x02\x03")

def test_errors(self):

self.assertRaises(UnicodeDecodeError, codecs.utf_32_be_decode,

b"\xff", "strict", True)

def test_issue8941(self):

# Issue #8941: insufficient result allocation when decoding into

# surrogate pairs on UCS-2 builds.

encoded = b'\x00\x01\x00\x00' * 1024

self.assertEqual('\U00010000' * 1024,

encoding = "utf-16"

if sys.byteorder == 'little':

ill_formed_sequence = b"\x80\xdc"

else:

ill_formed_sequence = b"\xdc\x80"

spamle = b'\xff\xfes\x00p\x00a\x00m\x00s\x00p\x00a\x00m\x00'

spambe = b'\xfe\xff\x00s\x00p\x00a\x00m\x00s\x00p\x00a\x00m'

def test_only_one_bom(self):

_,_,reader,writer = codecs.lookup(self.encoding)

# encode some stream

s = io.BytesIO()

f = writer(s)

f.write("spam")

f.write("spam")

d = s.getvalue()

# check whether there is exactly one BOM in it

self.assertTrue(d == self.spamle or d == self.spambe)

# try to read it back

s = io.BytesIO(d)

f = reader(s)

self.assertEqual(f.read(), "spamspam")

def test_badbom(self):

s = io.BytesIO(b"\xff\xff")

f = codecs.getreader(self.encoding)(s)

self.assertRaises(UnicodeError, f.read)

s = io.BytesIO(b"\xff\xff\xff\xff")

f = codecs.getreader(self.encoding)(s)

self.assertRaises(UnicodeError, f.read)

def test_partial(self):

self.check_partial(

"\x00\xff\u0100\uffff\U00010000",

[

"", # first byte of BOM read

"", # second byte of BOM read => byteorder known

"",

"\x00",

"\x00",

"\x00\xff",

"\x00\xff",

"\x00\xff\u0100",

"\x00\xff\u0100",

"\x00\xff\u0100\uffff",

"\x00\xff\u0100\uffff",

"\x00\xff\u0100\uffff",

"\x00\xff\u0100\uffff",

"\x00\xff\u0100\uffff\U00010000",

]

)

def test_handlers(self):

self.assertEqual(('\ufffd', 1),

codecs.utf_16_decode(b'\x01', 'replace', True))

self.assertEqual(('', 1),

codecs.utf_16_decode(b'\x01', 'ignore', True))

def test_errors(self):

self.assertRaises(UnicodeDecodeError, codecs.utf_16_decode,

b"\xff", "strict", True)

def test_decoder_state(self):

self.check_state_handling_decode(self.encoding,

"spamspam", self.spamle)

self.check_state_handling_decode(self.encoding,

"spamspam", self.spambe)

def test_bug691291(self):

# Files are always opened in binary mode, even if no binary mode was

# specified. This means that no automatic conversion of '\n' is done

# on reading and writing.

s1 = 'Hello\r\nworld\r\n'

s = s1.encode(self.encoding)

self.addCleanup(os_helper.unlink, os_helper.TESTFN)

with open(os_helper.TESTFN, 'wb') as fp:

fp.write(s)

with warnings_helper.check_warnings(('', DeprecationWarning)):

reader = codecs.open(os_helper.TESTFN, 'U', encoding=self.encoding)

with reader:

encoding = "utf-16-le"

ill_formed_sequence = b"\x80\xdc"

def test_partial(self):

self.check_partial(

"\x00\xff\u0100\uffff\U00010000",

[

"",

"\x00",

"\x00",

"\x00\xff",

"\x00\xff",

"\x00\xff\u0100",

"\x00\xff\u0100",

"\x00\xff\u0100\uffff",

"\x00\xff\u0100\uffff",

"\x00\xff\u0100\uffff",

"\x00\xff\u0100\uffff",

"\x00\xff\u0100\uffff\U00010000",

]

)

def test_errors(self):

tests = [

(b'\xff', '\ufffd'),

(b'A\x00Z', 'A\ufffd'),

(b'A\x00B\x00C\x00D\x00Z', 'ABCD\ufffd'),

(b'\x00\xd8', '\ufffd'),

(b'\x00\xd8A', '\ufffd'),

(b'\x00\xd8A\x00', '\ufffdA'),

(b'\x00\xdcA\x00', '\ufffdA'),

]

for raw, expected in tests:

self.assertRaises(UnicodeDecodeError, codecs.utf_16_le_decode,

raw, 'strict', True)

self.assertEqual(raw.decode('utf-16le', 'replace'), expected)

def test_nonbmp(self):

self.assertEqual("\U00010203".encode(self.encoding),

b'\x00\xd8\x03\xde')

self.assertEqual(b'\x00\xd8\x03\xde'.decode(self.encoding),

encoding = "utf-16-be"

ill_formed_sequence = b"\xdc\x80"

def test_partial(self):

self.check_partial(

"\x00\xff\u0100\uffff\U00010000",

[

"",

"\x00",

"\x00",

"\x00\xff",

"\x00\xff",

"\x00\xff\u0100",

"\x00\xff\u0100",

"\x00\xff\u0100\uffff",

"\x00\xff\u0100\uffff",

"\x00\xff\u0100\uffff",

"\x00\xff\u0100\uffff",

"\x00\xff\u0100\uffff\U00010000",

]

)

def test_errors(self):

tests = [

(b'\xff', '\ufffd'),

(b'\x00A\xff', 'A\ufffd'),

(b'\x00A\x00B\x00C\x00DZ', 'ABCD\ufffd'),

(b'\xd8\x00', '\ufffd'),

(b'\xd8\x00\xdc', '\ufffd'),

(b'\xd8\x00\x00A', '\ufffdA'),

(b'\xdc\x00\x00A', '\ufffdA'),

]

for raw, expected in tests:

self.assertRaises(UnicodeDecodeError, codecs.utf_16_be_decode,

raw, 'strict', True)

self.assertEqual(raw.decode('utf-16be', 'replace'), expected)

def test_nonbmp(self):

self.assertEqual("\U00010203".encode(self.encoding),

b'\xd8\x00\xde\x03')

self.assertEqual(b'\xd8\x00\xde\x03'.decode(self.encoding),

encoding = "utf-8"

ill_formed_sequence = b"\xed\xb2\x80"

ill_formed_sequence_replace = "\ufffd" * 3

BOM = b''

def test_partial(self):

self.check_partial(

"\x00\xff\u07ff\u0800\uffff\U00010000",

[

"\x00",

"\x00",

"\x00\xff",

"\x00\xff",

"\x00\xff\u07ff",

"\x00\xff\u07ff",

"\x00\xff\u07ff",

"\x00\xff\u07ff\u0800",

"\x00\xff\u07ff\u0800",

"\x00\xff\u07ff\u0800",

"\x00\xff\u07ff\u0800\uffff",

"\x00\xff\u07ff\u0800\uffff",

"\x00\xff\u07ff\u0800\uffff",

"\x00\xff\u07ff\u0800\uffff",

"\x00\xff\u07ff\u0800\uffff\U00010000",

]

)

def test_decoder_state(self):

u = "\x00\x7f\x80\xff\u0100\u07ff\u0800\uffff\U0010ffff"

self.check_state_handling_decode(self.encoding,

u, u.encode(self.encoding))

def test_decode_error(self):

for data, error_handler, expected in (

(b'[\x80\xff]', 'ignore', '[]'),

(b'[\x80\xff]', 'replace', '[\ufffd\ufffd]'),

(b'[\x80\xff]', 'surrogateescape', '[\udc80\udcff]'),

(b'[\x80\xff]', 'backslashreplace', '[\\x80\\xff]'),

):

with self.subTest(data=data, error_handler=error_handler,

expected=expected):

self.assertEqual(data.decode(self.encoding, error_handler),

expected)

def test_lone_surrogates(self):

super().test_lone_surrogates()

# not sure if this is making sense for

# UTF-16 and UTF-32

self.assertEqual("[\uDC80]".encode(self.encoding, "surrogateescape"),

self.BOM + b'[\x80]')

with self.assertRaises(UnicodeEncodeError) as cm:

"[\uDC80\uD800\uDFFF]".encode(self.encoding, "surrogateescape")

exc = cm.exception

self.assertEqual(exc.object[exc.start:exc.end], '\uD800\uDFFF')

def test_surrogatepass_handler(self):

self.assertEqual("abc\ud800def".encode(self.encoding, "surrogatepass"),

self.BOM + b"abc\xed\xa0\x80def")

self.assertEqual("\U00010fff\uD800".encode(self.encoding, "surrogatepass"),

self.BOM + b"\xf0\x90\xbf\xbf\xed\xa0\x80")

self.assertEqual("[\uD800\uDC80]".encode(self.encoding, "surrogatepass"),

self.BOM + b'[\xed\xa0\x80\xed\xb2\x80]')

self.assertEqual(b"abc\xed\xa0\x80def".decode(self.encoding, "surrogatepass"),

"abc\ud800def")

self.assertEqual(b"\xf0\x90\xbf\xbf\xed\xa0\x80".decode(self.encoding, "surrogatepass"),

"\U00010fff\uD800")

self.assertTrue(codecs.lookup_error("surrogatepass"))

with self.assertRaises(UnicodeDecodeError):

b"abc\xed\xa0".decode(self.encoding, "surrogatepass")

with self.assertRaises(UnicodeDecodeError):

b"abc\xed\xa0z".decode(self.encoding, "surrogatepass")

def test_incremental_errors(self):

# Test that the incremental decoder can fail with final=False.

# See issue #24214

cases = [b'\x80', b'\xBF', b'\xC0', b'\xC1', b'\xF5', b'\xF6', b'\xFF']

for prefix in (b'\xC2', b'\xDF', b'\xE0', b'\xE0\xA0', b'\xEF',

b'\xEF\xBF', b'\xF0', b'\xF0\x90', b'\xF0\x90\x80',

b'\xF4', b'\xF4\x8F', b'\xF4\x8F\xBF'):

for suffix in b'\x7F', b'\xC0':

cases.append(prefix + suffix)

cases.extend((b'\xE0\x80', b'\xE0\x9F', b'\xED\xA0\x80',

b'\xED\xBF\xBF', b'\xF0\x80', b'\xF0\x8F', b'\xF4\x90'))

for data in cases:

with self.subTest(data=data):

dec = codecs.getincrementaldecoder(self.encoding)()

encoding = "utf-7"

def test_ascii(self):

# Set D (directly encoded characters)

set_d = ('ABCDEFGHIJKLMNOPQRSTUVWXYZ'

'abcdefghijklmnopqrstuvwxyz'

'0123456789'

'\'(),-./:?')

self.assertEqual(set_d.encode(self.encoding), set_d.encode('ascii'))

self.assertEqual(set_d.encode('ascii').decode(self.encoding), set_d)

# Set O (optional direct characters)

set_o = ' !"#$%&*;<=>@[]^_`{|}'

self.assertEqual(set_o.encode(self.encoding), set_o.encode('ascii'))

self.assertEqual(set_o.encode('ascii').decode(self.encoding), set_o)

# +

self.assertEqual('a+b'.encode(self.encoding), b'a+-b')

self.assertEqual(b'a+-b'.decode(self.encoding), 'a+b')

# White spaces

ws = ' \t\n\r'

self.assertEqual(ws.encode(self.encoding), ws.encode('ascii'))

self.assertEqual(ws.encode('ascii').decode(self.encoding), ws)

# Other ASCII characters

other_ascii = ''.join(sorted(set(bytes(range(0x80)).decode()) -

set(set_d + set_o + '+' + ws)))

self.assertEqual(other_ascii.encode(self.encoding),

b'+AAAAAQACAAMABAAFAAYABwAIAAsADAAOAA8AEAARABIAEwAU'

b'ABUAFgAXABgAGQAaABsAHAAdAB4AHwBcAH4Afw-')

def test_partial(self):

self.check_partial(

'a+-b\x00c\x80d\u0100e\U00010000f',

[

'a',