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