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"

@unittest.expectedFailure # TODO: RUSTPYTHON

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')

@unittest.expectedFailure # TODO: RUSTPYTHON

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")

@unittest.expectedFailure # TODO: RUSTPYTHON

def test_badbom(self):

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

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

self.assertRaises(UnicodeDecodeError, f.read)

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

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

self.assertRaises(UnicodeDecodeError, f.read)

@unittest.expectedFailure # TODO: RUSTPYTHON

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",

]

)

@unittest.expectedFailure # TODO: RUSTPYTHON

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))

@unittest.expectedFailure # TODO: RUSTPYTHON

def test_errors(self):

self.assertRaises(UnicodeDecodeError, codecs.utf_32_decode,

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

@unittest.expectedFailure # TODO: RUSTPYTHON

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)

@unittest.expectedFailure # TODO: RUSTPYTHON

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,

codecs.utf_32_decode(encoded_be)[0])

@unittest.expectedFailure # TODO: RUSTPYTHON

def test_bug1098990_a(self):

return super().test_bug1098990_a()

@unittest.expectedFailure # TODO: RUSTPYTHON

def test_bug1098990_b(self):

return super().test_bug1098990_b()

@unittest.expectedFailure # TODO: RUSTPYTHON

def test_bug1175396(self):

return super().test_bug1175396()

@unittest.expectedFailure # TODO: RUSTPYTHON

def test_incremental_surrogatepass(self):

return super().test_incremental_surrogatepass()

@unittest.expectedFailure # TODO: RUSTPYTHON

def test_mixed_readline_and_read(self):

return super().test_mixed_readline_and_read()

@unittest.expectedFailure # TODO: RUSTPYTHON

def test_readline(self):

return super().test_readline()

@unittest.expectedFailure # TODO: RUSTPYTHON

def test_readlinequeue(self):

encoding = "utf-32-le"

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

@unittest.expectedFailure # TODO: RUSTPYTHON

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",

]

)

@unittest.expectedFailure # TODO: RUSTPYTHON

def test_simple(self):

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

@unittest.expectedFailure # TODO: RUSTPYTHON

def test_errors(self):

self.assertRaises(UnicodeDecodeError, codecs.utf_32_le_decode,

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

@unittest.expectedFailure # TODO: RUSTPYTHON

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,

codecs.utf_32_le_decode(encoded)[0])

@unittest.expectedFailure # TODO: RUSTPYTHON

def test_bug1098990_a(self):

return super().test_bug1098990_a()

@unittest.expectedFailure # TODO: RUSTPYTHON

def test_bug1098990_b(self):

return super().test_bug1098990_b()

@unittest.expectedFailure # TODO: RUSTPYTHON

def test_bug1175396(self):

return super().test_bug1175396()

@unittest.expectedFailure # TODO: RUSTPYTHON

def test_incremental_surrogatepass(self):

return super().test_incremental_surrogatepass()

@unittest.expectedFailure # TODO: RUSTPYTHON

def test_mixed_readline_and_read(self):

return super().test_mixed_readline_and_read()

@unittest.expectedFailure # TODO: RUSTPYTHON

def test_readline(self):

return super().test_readline()

@unittest.expectedFailure # TODO: RUSTPYTHON

def test_readlinequeue(self):

encoding = "utf-32-be"

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

@unittest.expectedFailure # TODO: RUSTPYTHON

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",

]

)

@unittest.expectedFailure # TODO: RUSTPYTHON

def test_simple(self):

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

@unittest.expectedFailure # TODO: RUSTPYTHON

def test_errors(self):

self.assertRaises(UnicodeDecodeError, codecs.utf_32_be_decode,

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

@unittest.expectedFailure # TODO: RUSTPYTHON

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,

codecs.utf_32_be_decode(encoded)[0])

@unittest.expectedFailure # TODO: RUSTPYTHON

def test_bug1098990_a(self):

return super().test_bug1098990_a()

@unittest.expectedFailure # TODO: RUSTPYTHON

def test_bug1098990_b(self):

return super().test_bug1098990_b()

@unittest.expectedFailure # TODO: RUSTPYTHON

def test_bug1175396(self):

return super().test_bug1175396()

@unittest.expectedFailure # TODO: RUSTPYTHON

def test_incremental_surrogatepass(self):

return super().test_incremental_surrogatepass()

@unittest.expectedFailure # TODO: RUSTPYTHON

def test_mixed_readline_and_read(self):

return super().test_mixed_readline_and_read()

@unittest.expectedFailure # TODO: RUSTPYTHON

def test_readline(self):

return super().test_readline()

@unittest.expectedFailure # TODO: RUSTPYTHON

def test_readlinequeue(self):

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(UnicodeDecodeError, f.read)

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

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

self.assertRaises(UnicodeDecodeError, f.read)

@unittest.expectedFailure # TODO: RUSTPYTHON

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",

]

)

@unittest.expectedFailure # TODO: RUSTPYTHON

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):

# If encoding is not None, then

# 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 codecs.open(os_helper.TESTFN, 'r',

encoding=self.encoding) as reader:

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

def test_invalid_modes(self):

for mode in ('U', 'rU', 'r+U'):

with self.assertRaises(ValueError) as cm:

codecs.open(os_helper.TESTFN, mode, encoding=self.encoding)

self.assertIn('invalid mode', str(cm.exception))

for mode in ('rt', 'wt', 'at', 'r+t'):

with self.assertRaises(ValueError) as cm:

codecs.open(os_helper.TESTFN, mode, encoding=self.encoding)

self.assertIn("can't have text and binary mode at once",

str(cm.exception))

@unittest.expectedFailure # TODO: RUSTPYTHON

def test_incremental_surrogatepass(self):

encoding = "utf-16-le"

ill_formed_sequence = b"\x80\xdc"

@unittest.expectedFailure # TODO: RUSTPYTHON

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",

]

)

@unittest.expectedFailure # TODO: RUSTPYTHON

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),

"\U00010203")

@unittest.expectedFailure # TODO: RUSTPYTHON

def test_incremental_surrogatepass(self):

encoding = "utf-16-be"

ill_formed_sequence = b"\xdc\x80"

@unittest.expectedFailure # TODO: RUSTPYTHON

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",

]

)

@unittest.expectedFailure # TODO: RUSTPYTHON

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),

"\U00010203")

@unittest.expectedFailure # TODO: RUSTPYTHON

def test_incremental_surrogatepass(self):

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,