1#
2# multibytecodec_support.py
3#   Common Unittest Routines for CJK codecs
4#
5
6import codecs
7import os
8import re
9import sys
10import unittest
11from http.client import HTTPException
12from test import support
13from io import BytesIO
14
15class TestBase:
16    encoding        = ''   # codec name
17    codec           = None # codec tuple (with 4 elements)
18    tstring         = None # must set. 2 strings to test StreamReader
19
20    codectests      = None # must set. codec test tuple
21    roundtriptest   = 1    # set if roundtrip is possible with unicode
22    has_iso10646    = 0    # set if this encoding contains whole iso10646 map
23    xmlcharnametest = None # string to test xmlcharrefreplace
24    unmappedunicode = '\udeee' # a unicode code point that is not mapped.
25
26    def setUp(self):
27        if self.codec is None:
28            self.codec = codecs.lookup(self.encoding)
29        self.encode = self.codec.encode
30        self.decode = self.codec.decode
31        self.reader = self.codec.streamreader
32        self.writer = self.codec.streamwriter
33        self.incrementalencoder = self.codec.incrementalencoder
34        self.incrementaldecoder = self.codec.incrementaldecoder
35
36    def test_chunkcoding(self):
37        tstring_lines = []
38        for b in self.tstring:
39            lines = b.split(b"\n")
40            last = lines.pop()
41            assert last == b""
42            lines = [line + b"\n" for line in lines]
43            tstring_lines.append(lines)
44        for native, utf8 in zip(*tstring_lines):
45            u = self.decode(native)[0]
46            self.assertEqual(u, utf8.decode('utf-8'))
47            if self.roundtriptest:
48                self.assertEqual(native, self.encode(u)[0])
49
50    def test_errorhandle(self):
51        for source, scheme, expected in self.codectests:
52            if isinstance(source, bytes):
53                func = self.decode
54            else:
55                func = self.encode
56            if expected:
57                result = func(source, scheme)[0]
58                if func is self.decode:
59                    self.assertTrue(type(result) is str, type(result))
60                    self.assertEqual(result, expected,
61                                     '%a.decode(%r, %r)=%a != %a'
62                                     % (source, self.encoding, scheme, result,
63                                        expected))
64                else:
65                    self.assertTrue(type(result) is bytes, type(result))
66                    self.assertEqual(result, expected,
67                                     '%a.encode(%r, %r)=%a != %a'
68                                     % (source, self.encoding, scheme, result,
69                                        expected))
70            else:
71                self.assertRaises(UnicodeError, func, source, scheme)
72
73    def test_xmlcharrefreplace(self):
74        if self.has_iso10646:
75            self.skipTest('encoding contains full ISO 10646 map')
76
77        s = "\u0b13\u0b23\u0b60 nd eggs"
78        self.assertEqual(
79            self.encode(s, "xmlcharrefreplace")[0],
80            b"ଓଣୠ nd eggs"
81        )
82
83    def test_customreplace_encode(self):
84        if self.has_iso10646:
85            self.skipTest('encoding contains full ISO 10646 map')
86
87        from html.entities import codepoint2name
88
89        def xmlcharnamereplace(exc):
90            if not isinstance(exc, UnicodeEncodeError):
91                raise TypeError("don't know how to handle %r" % exc)
92            l = []
93            for c in exc.object[exc.start:exc.end]:
94                if ord(c) in codepoint2name:
95                    l.append("&%s;" % codepoint2name[ord(c)])
96                else:
97                    l.append("&#%d;" % ord(c))
98            return ("".join(l), exc.end)
99
100        codecs.register_error("test.xmlcharnamereplace", xmlcharnamereplace)
101
102        if self.xmlcharnametest:
103            sin, sout = self.xmlcharnametest
104        else:
105            sin = "\xab\u211c\xbb = \u2329\u1234\u232a"
106            sout = b"«ℜ» = ⟨ሴ⟩"
107        self.assertEqual(self.encode(sin,
108                                    "test.xmlcharnamereplace")[0], sout)
109
110    def test_callback_returns_bytes(self):
111        def myreplace(exc):
112            return (b"1234", exc.end)
113        codecs.register_error("test.cjktest", myreplace)
114        enc = self.encode("abc" + self.unmappedunicode + "def", "test.cjktest")[0]
115        self.assertEqual(enc, b"abc1234def")
116
117    def test_callback_wrong_objects(self):
118        def myreplace(exc):
119            return (ret, exc.end)
120        codecs.register_error("test.cjktest", myreplace)
121
122        for ret in ([1, 2, 3], [], None, object()):
123            self.assertRaises(TypeError, self.encode, self.unmappedunicode,
124                              'test.cjktest')
125
126    def test_callback_long_index(self):
127        def myreplace(exc):
128            return ('x', int(exc.end))
129        codecs.register_error("test.cjktest", myreplace)
130        self.assertEqual(self.encode('abcd' + self.unmappedunicode + 'efgh',
131                                     'test.cjktest'), (b'abcdxefgh', 9))
132
133        def myreplace(exc):
134            return ('x', sys.maxsize + 1)
135        codecs.register_error("test.cjktest", myreplace)
136        self.assertRaises(IndexError, self.encode, self.unmappedunicode,
137                          'test.cjktest')
138
139    def test_callback_None_index(self):
140        def myreplace(exc):
141            return ('x', None)
142        codecs.register_error("test.cjktest", myreplace)
143        self.assertRaises(TypeError, self.encode, self.unmappedunicode,
144                          'test.cjktest')
145
146    def test_callback_backward_index(self):
147        def myreplace(exc):
148            if myreplace.limit > 0:
149                myreplace.limit -= 1
150                return ('REPLACED', 0)
151            else:
152                return ('TERMINAL', exc.end)
153        myreplace.limit = 3
154        codecs.register_error("test.cjktest", myreplace)
155        self.assertEqual(self.encode('abcd' + self.unmappedunicode + 'efgh',
156                                     'test.cjktest'),
157                (b'abcdREPLACEDabcdREPLACEDabcdREPLACEDabcdTERMINALefgh', 9))
158
159    def test_callback_forward_index(self):
160        def myreplace(exc):
161            return ('REPLACED', exc.end + 2)
162        codecs.register_error("test.cjktest", myreplace)
163        self.assertEqual(self.encode('abcd' + self.unmappedunicode + 'efgh',
164                                     'test.cjktest'), (b'abcdREPLACEDgh', 9))
165
166    def test_callback_index_outofbound(self):
167        def myreplace(exc):
168            return ('TERM', 100)
169        codecs.register_error("test.cjktest", myreplace)
170        self.assertRaises(IndexError, self.encode, self.unmappedunicode,
171                          'test.cjktest')
172
173    def test_incrementalencoder(self):
174        UTF8Reader = codecs.getreader('utf-8')
175        for sizehint in [None] + list(range(1, 33)) + \
176                        [64, 128, 256, 512, 1024]:
177            istream = UTF8Reader(BytesIO(self.tstring[1]))
178            ostream = BytesIO()
179            encoder = self.incrementalencoder()
180            while 1:
181                if sizehint is not None:
182                    data = istream.read(sizehint)
183                else:
184                    data = istream.read()
185
186                if not data:
187                    break
188                e = encoder.encode(data)
189                ostream.write(e)
190
191            self.assertEqual(ostream.getvalue(), self.tstring[0])
192
193    def test_incrementaldecoder(self):
194        UTF8Writer = codecs.getwriter('utf-8')
195        for sizehint in [None, -1] + list(range(1, 33)) + \
196                        [64, 128, 256, 512, 1024]:
197            istream = BytesIO(self.tstring[0])
198            ostream = UTF8Writer(BytesIO())
199            decoder = self.incrementaldecoder()
200            while 1:
201                data = istream.read(sizehint)
202                if not data:
203                    break
204                else:
205                    u = decoder.decode(data)
206                    ostream.write(u)
207
208            self.assertEqual(ostream.getvalue(), self.tstring[1])
209
210    def test_incrementalencoder_error_callback(self):
211        inv = self.unmappedunicode
212
213        e = self.incrementalencoder()
214        self.assertRaises(UnicodeEncodeError, e.encode, inv, True)
215
216        e.errors = 'ignore'
217        self.assertEqual(e.encode(inv, True), b'')
218
219        e.reset()
220        def tempreplace(exc):
221            return ('called', exc.end)
222        codecs.register_error('test.incremental_error_callback', tempreplace)
223        e.errors = 'test.incremental_error_callback'
224        self.assertEqual(e.encode(inv, True), b'called')
225
226        # again
227        e.errors = 'ignore'
228        self.assertEqual(e.encode(inv, True), b'')
229
230    def test_streamreader(self):
231        UTF8Writer = codecs.getwriter('utf-8')
232        for name in ["read", "readline", "readlines"]:
233            for sizehint in [None, -1] + list(range(1, 33)) + \
234                            [64, 128, 256, 512, 1024]:
235                istream = self.reader(BytesIO(self.tstring[0]))
236                ostream = UTF8Writer(BytesIO())
237                func = getattr(istream, name)
238                while 1:
239                    data = func(sizehint)
240                    if not data:
241                        break
242                    if name == "readlines":
243                        ostream.writelines(data)
244                    else:
245                        ostream.write(data)
246
247                self.assertEqual(ostream.getvalue(), self.tstring[1])
248
249    def test_streamwriter(self):
250        readfuncs = ('read', 'readline', 'readlines')
251        UTF8Reader = codecs.getreader('utf-8')
252        for name in readfuncs:
253            for sizehint in [None] + list(range(1, 33)) + \
254                            [64, 128, 256, 512, 1024]:
255                istream = UTF8Reader(BytesIO(self.tstring[1]))
256                ostream = self.writer(BytesIO())
257                func = getattr(istream, name)
258                while 1:
259                    if sizehint is not None:
260                        data = func(sizehint)
261                    else:
262                        data = func()
263
264                    if not data:
265                        break
266                    if name == "readlines":
267                        ostream.writelines(data)
268                    else:
269                        ostream.write(data)
270
271                self.assertEqual(ostream.getvalue(), self.tstring[0])
272
273    def test_streamwriter_reset_no_pending(self):
274        # Issue #23247: Calling reset() on a fresh StreamWriter instance
275        # (without pending data) must not crash
276        stream = BytesIO()
277        writer = self.writer(stream)
278        writer.reset()
279
280
281class TestBase_Mapping(unittest.TestCase):
282    pass_enctest = []
283    pass_dectest = []
284    supmaps = []
285    codectests = []
286
287    def setUp(self):
288        try:
289            self.open_mapping_file().close() # test it to report the error early
290        except (OSError, HTTPException):
291            self.skipTest("Could not retrieve "+self.mapfileurl)
292
293    def open_mapping_file(self):
294        return support.open_urlresource(self.mapfileurl)
295
296    def test_mapping_file(self):
297        if self.mapfileurl.endswith('.xml'):
298            self._test_mapping_file_ucm()
299        else:
300            self._test_mapping_file_plain()
301
302    def _test_mapping_file_plain(self):
303        unichrs = lambda s: ''.join(map(chr, map(eval, s.split('+'))))
304        urt_wa = {}
305
306        with self.open_mapping_file() as f:
307            for line in f:
308                if not line:
309                    break
310                data = line.split('#')[0].strip().split()
311                if len(data) != 2:
312                    continue
313
314                csetval = eval(data[0])
315                if csetval <= 0x7F:
316                    csetch = bytes([csetval & 0xff])
317                elif csetval >= 0x1000000:
318                    csetch = bytes([(csetval >> 24), ((csetval >> 16) & 0xff),
319                                    ((csetval >> 8) & 0xff), (csetval & 0xff)])
320                elif csetval >= 0x10000:
321                    csetch = bytes([(csetval >> 16), ((csetval >> 8) & 0xff),
322                                    (csetval & 0xff)])
323                elif csetval >= 0x100:
324                    csetch = bytes([(csetval >> 8), (csetval & 0xff)])
325                else:
326                    continue
327
328                unich = unichrs(data[1])
329                if ord(unich) == 0xfffd or unich in urt_wa:
330                    continue
331                urt_wa[unich] = csetch
332
333                self._testpoint(csetch, unich)
334
335    def _test_mapping_file_ucm(self):
336        with self.open_mapping_file() as f:
337            ucmdata = f.read()
338        uc = re.findall('<a u="([A-F0-9]{4})" b="([0-9A-F ]+)"/>', ucmdata)
339        for uni, coded in uc:
340            unich = chr(int(uni, 16))
341            codech = bytes(int(c, 16) for c in coded.split())
342            self._testpoint(codech, unich)
343
344    def test_mapping_supplemental(self):
345        for mapping in self.supmaps:
346            self._testpoint(*mapping)
347
348    def _testpoint(self, csetch, unich):
349        if (csetch, unich) not in self.pass_enctest:
350            self.assertEqual(unich.encode(self.encoding), csetch)
351        if (csetch, unich) not in self.pass_dectest:
352            self.assertEqual(str(csetch, self.encoding), unich)
353
354    def test_errorhandle(self):
355        for source, scheme, expected in self.codectests:
356            if isinstance(source, bytes):
357                func = source.decode
358            else:
359                func = source.encode
360            if expected:
361                if isinstance(source, bytes):
362                    result = func(self.encoding, scheme)
363                    self.assertTrue(type(result) is str, type(result))
364                    self.assertEqual(result, expected,
365                                     '%a.decode(%r, %r)=%a != %a'
366                                     % (source, self.encoding, scheme, result,
367                                        expected))
368                else:
369                    result = func(self.encoding, scheme)
370                    self.assertTrue(type(result) is bytes, type(result))
371                    self.assertEqual(result, expected,
372                                     '%a.encode(%r, %r)=%a != %a'
373                                     % (source, self.encoding, scheme, result,
374                                        expected))
375            else:
376                self.assertRaises(UnicodeError, func, self.encoding, scheme)
377
378def load_teststring(name):
379    dir = os.path.join(os.path.dirname(__file__), 'cjkencodings')
380    with open(os.path.join(dir, name + '.txt'), 'rb') as f:
381        encoded = f.read()
382    with open(os.path.join(dir, name + '-utf8.txt'), 'rb') as f:
383        utf8 = f.read()
384    return encoded, utf8
385