|
1 | 1 | import _testcapi |
2 | 2 | import codecs |
| 3 | +import contextlib |
3 | 4 | import io |
4 | 5 | import locale |
5 | 6 | import sys |
@@ -2292,46 +2293,190 @@ class TransformCodecTest(unittest.TestCase): |
2292 | 2293 | def test_basics(self): |
2293 | 2294 | binput = bytes(range(256)) |
2294 | 2295 | for encoding in bytes_transform_encodings: |
2295 | | - # generic codecs interface |
2296 | | - (o, size) = codecs.getencoder(encoding)(binput) |
2297 | | - self.assertEqual(size, len(binput)) |
2298 | | - (i, size) = codecs.getdecoder(encoding)(o) |
2299 | | - self.assertEqual(size, len(o)) |
2300 | | - self.assertEqual(i, binput) |
| 2296 | + with self.subTest(encoding=encoding): |
| 2297 | + # generic codecs interface |
| 2298 | + (o, size) = codecs.getencoder(encoding)(binput) |
| 2299 | + self.assertEqual(size, len(binput)) |
| 2300 | + (i, size) = codecs.getdecoder(encoding)(o) |
| 2301 | + self.assertEqual(size, len(o)) |
| 2302 | + self.assertEqual(i, binput) |
2301 | 2303 |
|
2302 | 2304 | def test_read(self): |
2303 | 2305 | for encoding in bytes_transform_encodings: |
2304 | | - sin = codecs.encode(b"\x80", encoding) |
2305 | | - reader = codecs.getreader(encoding)(io.BytesIO(sin)) |
2306 | | - sout = reader.read() |
2307 | | - self.assertEqual(sout, b"\x80") |
| 2306 | + with self.subTest(encoding=encoding): |
| 2307 | + sin = codecs.encode(b"\x80", encoding) |
| 2308 | + reader = codecs.getreader(encoding)(io.BytesIO(sin)) |
| 2309 | + sout = reader.read() |
| 2310 | + self.assertEqual(sout, b"\x80") |
2308 | 2311 |
|
2309 | 2312 | def test_readline(self): |
2310 | 2313 | for encoding in bytes_transform_encodings: |
2311 | 2314 | if encoding in ['uu_codec', 'zlib_codec']: |
2312 | 2315 | continue |
2313 | | - sin = codecs.encode(b"\x80", encoding) |
2314 | | - reader = codecs.getreader(encoding)(io.BytesIO(sin)) |
2315 | | - sout = reader.readline() |
2316 | | - self.assertEqual(sout, b"\x80") |
| 2316 | + with self.subTest(encoding=encoding): |
| 2317 | + sin = codecs.encode(b"\x80", encoding) |
| 2318 | + reader = codecs.getreader(encoding)(io.BytesIO(sin)) |
| 2319 | + sout = reader.readline() |
| 2320 | + self.assertEqual(sout, b"\x80") |
2317 | 2321 |
|
2318 | 2322 | def test_buffer_api_usage(self): |
2319 | 2323 | # We check all the transform codecs accept memoryview input |
2320 | 2324 | # for encoding and decoding |
2321 | 2325 | # and also that they roundtrip correctly |
2322 | 2326 | original = b"12345\x80" |
2323 | 2327 | for encoding in bytes_transform_encodings: |
2324 | | - data = original |
2325 | | - view = memoryview(data) |
2326 | | - data = codecs.encode(data, encoding) |
2327 | | - view_encoded = codecs.encode(view, encoding) |
2328 | | - self.assertEqual(view_encoded, data) |
2329 | | - view = memoryview(data) |
2330 | | - data = codecs.decode(data, encoding) |
2331 | | - self.assertEqual(data, original) |
2332 | | - view_decoded = codecs.decode(view, encoding) |
2333 | | - self.assertEqual(view_decoded, data) |
| 2328 | + with self.subTest(encoding=encoding): |
| 2329 | + data = original |
| 2330 | + view = memoryview(data) |
| 2331 | + data = codecs.encode(data, encoding) |
| 2332 | + view_encoded = codecs.encode(view, encoding) |
| 2333 | + self.assertEqual(view_encoded, data) |
| 2334 | + view = memoryview(data) |
| 2335 | + data = codecs.decode(data, encoding) |
| 2336 | + self.assertEqual(data, original) |
| 2337 | + view_decoded = codecs.decode(view, encoding) |
| 2338 | + self.assertEqual(view_decoded, data) |
| 2339 | + |
| 2340 | + def test_type_error_for_text_input(self): |
| 2341 | + # Check binary -> binary codecs give a good error for str input |
| 2342 | + bad_input = "bad input type" |
| 2343 | + for encoding in bytes_transform_encodings: |
| 2344 | + with self.subTest(encoding=encoding): |
| 2345 | + msg = "^encoding with '{}' codec failed".format(encoding) |
| 2346 | + with self.assertRaisesRegex(TypeError, msg) as failure: |
| 2347 | + bad_input.encode(encoding) |
| 2348 | + self.assertTrue(isinstance(failure.exception.__cause__, |
| 2349 | + TypeError)) |
| 2350 | + |
| 2351 | + def test_type_error_for_binary_input(self): |
| 2352 | + # Check str -> str codec gives a good error for binary input |
| 2353 | + for bad_input in (b"immutable", bytearray(b"mutable")): |
| 2354 | + with self.subTest(bad_input=bad_input): |
| 2355 | + msg = "^decoding with 'rot_13' codec failed" |
| 2356 | + with self.assertRaisesRegex(AttributeError, msg) as failure: |
| 2357 | + bad_input.decode("rot_13") |
| 2358 | + self.assertTrue(isinstance(failure.exception.__cause__, |
| 2359 | + AttributeError)) |
| 2360 | + |
| 2361 | + def test_bad_decoding_output_type(self): |
| 2362 | + # Check bytes.decode and bytearray.decode give a good error |
| 2363 | + # message for binary -> binary codecs |
| 2364 | + data = b"encode first to ensure we meet any format restrictions" |
| 2365 | + for encoding in bytes_transform_encodings: |
| 2366 | + with self.subTest(encoding=encoding): |
| 2367 | + encoded_data = codecs.encode(data, encoding) |
| 2368 | + fmt = ("'{}' decoder returned 'bytes' instead of 'str'; " |
| 2369 | + "use codecs.decode\(\) to decode to arbitrary types") |
| 2370 | + msg = fmt.format(encoding) |
| 2371 | + with self.assertRaisesRegex(TypeError, msg): |
| 2372 | + encoded_data.decode(encoding) |
| 2373 | + with self.assertRaisesRegex(TypeError, msg): |
| 2374 | + bytearray(encoded_data).decode(encoding) |
| 2375 | + |
| 2376 | + def test_bad_encoding_output_type(self): |
| 2377 | + # Check str.encode gives a good error message for str -> str codecs |
| 2378 | + msg = ("'rot_13' encoder returned 'str' instead of 'bytes'; " |
| 2379 | + "use codecs.encode\(\) to encode to arbitrary types") |
| 2380 | + with self.assertRaisesRegex(TypeError, msg): |
| 2381 | + "just an example message".encode("rot_13") |
| 2382 | + |
| 2383 | + |
| 2384 | +# The codec system tries to wrap exceptions in order to ensure the error |
| 2385 | +# mentions the operation being performed and the codec involved. We |
| 2386 | +# currently *only* want this to happen for relatively stateless |
| 2387 | +# exceptions, where the only significant information they contain is their |
| 2388 | +# type and a single str argument. |
| 2389 | +class ExceptionChainingTest(unittest.TestCase): |
2334 | 2390 |
|
| 2391 | + def setUp(self): |
| 2392 | + # There's no way to unregister a codec search function, so we just |
| 2393 | + # ensure we render this one fairly harmless after the test |
| 2394 | + # case finishes by using the test case repr as the codec name |
| 2395 | + # The codecs module normalizes codec names, although this doesn't |
| 2396 | + # appear to be formally documented... |
| 2397 | + self.codec_name = repr(self).lower().replace(" ", "-") |
| 2398 | + self.codec_info = None |
| 2399 | + codecs.register(self.get_codec) |
| 2400 | + |
| 2401 | + def get_codec(self, codec_name): |
| 2402 | + if codec_name != self.codec_name: |
| 2403 | + return None |
| 2404 | + return self.codec_info |
| 2405 | + |
| 2406 | + def set_codec(self, obj_to_raise): |
| 2407 | + def raise_obj(*args, **kwds): |
| 2408 | + raise obj_to_raise |
| 2409 | + self.codec_info = codecs.CodecInfo(raise_obj, raise_obj, |
| 2410 | + name=self.codec_name) |
| 2411 | + |
| 2412 | + @contextlib.contextmanager |
| 2413 | + def assertWrapped(self, operation, exc_type, msg): |
| 2414 | + full_msg = "{} with '{}' codec failed \({}: {}\)".format( |
| 2415 | + operation, self.codec_name, exc_type.__name__, msg) |
| 2416 | + with self.assertRaisesRegex(exc_type, full_msg) as caught: |
| 2417 | + yield caught |
| 2418 | + |
| 2419 | + def check_wrapped(self, obj_to_raise, msg): |
| 2420 | + self.set_codec(obj_to_raise) |
| 2421 | + with self.assertWrapped("encoding", RuntimeError, msg): |
| 2422 | + "str_input".encode(self.codec_name) |
| 2423 | + with self.assertWrapped("encoding", RuntimeError, msg): |
| 2424 | + codecs.encode("str_input", self.codec_name) |
| 2425 | + with self.assertWrapped("decoding", RuntimeError, msg): |
| 2426 | + b"bytes input".decode(self.codec_name) |
| 2427 | + with self.assertWrapped("decoding", RuntimeError, msg): |
| 2428 | + codecs.decode(b"bytes input", self.codec_name) |
| 2429 | + |
| 2430 | + def test_raise_by_type(self): |
| 2431 | + self.check_wrapped(RuntimeError, "") |
| 2432 | + |
| 2433 | + def test_raise_by_value(self): |
| 2434 | + msg = "This should be wrapped" |
| 2435 | + self.check_wrapped(RuntimeError(msg), msg) |
| 2436 | + |
| 2437 | + @contextlib.contextmanager |
| 2438 | + def assertNotWrapped(self, operation, exc_type, msg): |
| 2439 | + with self.assertRaisesRegex(exc_type, msg) as caught: |
| 2440 | + yield caught |
| 2441 | + actual_msg = str(caught.exception) |
| 2442 | + self.assertNotIn(operation, actual_msg) |
| 2443 | + self.assertNotIn(self.codec_name, actual_msg) |
| 2444 | + |
| 2445 | + def check_not_wrapped(self, obj_to_raise, msg): |
| 2446 | + self.set_codec(obj_to_raise) |
| 2447 | + with self.assertNotWrapped("encoding", RuntimeError, msg): |
| 2448 | + "str input".encode(self.codec_name) |
| 2449 | + with self.assertNotWrapped("encoding", RuntimeError, msg): |
| 2450 | + codecs.encode("str input", self.codec_name) |
| 2451 | + with self.assertNotWrapped("decoding", RuntimeError, msg): |
| 2452 | + b"bytes input".decode(self.codec_name) |
| 2453 | + with self.assertNotWrapped("decoding", RuntimeError, msg): |
| 2454 | + codecs.decode(b"bytes input", self.codec_name) |
| 2455 | + |
| 2456 | + def test_init_override_is_not_wrapped(self): |
| 2457 | + class CustomInit(RuntimeError): |
| 2458 | + def __init__(self): |
| 2459 | + pass |
| 2460 | + self.check_not_wrapped(CustomInit, "") |
| 2461 | + |
| 2462 | + def test_new_override_is_not_wrapped(self): |
| 2463 | + class CustomNew(RuntimeError): |
| 2464 | + def __new__(cls): |
| 2465 | + return super().__new__(cls) |
| 2466 | + self.check_not_wrapped(CustomNew, "") |
| 2467 | + |
| 2468 | + def test_instance_attribute_is_not_wrapped(self): |
| 2469 | + msg = "This should NOT be wrapped" |
| 2470 | + exc = RuntimeError(msg) |
| 2471 | + exc.attr = 1 |
| 2472 | + self.check_not_wrapped(exc, msg) |
| 2473 | + |
| 2474 | + def test_non_str_arg_is_not_wrapped(self): |
| 2475 | + self.check_not_wrapped(RuntimeError(1), "1") |
| 2476 | + |
| 2477 | + def test_multiple_args_is_not_wrapped(self): |
| 2478 | + msg = "\('a', 'b', 'c'\)" |
| 2479 | + self.check_not_wrapped(RuntimeError('a', 'b', 'c'), msg) |
2335 | 2480 |
|
2336 | 2481 |
|
2337 | 2482 | @unittest.skipUnless(sys.platform == 'win32', |
|
0 commit comments