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