1 # 2 # multibytecodec_support.py 3 # Common Unittest Routines for CJK codecs 4 # 5 6 import codecs 7 import os 8 import re 9 import sys 10 import unittest 11 from http.client import HTTPException 12 from test import support 13 from io import BytesIO 14 15 class 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 281 class 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 378 def 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
