|
1 | 1 | """Test the binascii C module.""" |
2 | 2 |
|
| 3 | +import sys |
| 4 | +import threading |
3 | 5 | import unittest |
4 | 6 | import binascii |
5 | 7 | import array |
6 | 8 | import re |
| 9 | +from test import support |
7 | 10 | from test.support import bigmemtest, _1G, _4G |
8 | 11 | from test.support.hypothesis_helper import hypothesis |
| 12 | +from test.support import threading_helper |
| 13 | +from test.support.script_helper import assert_python_ok |
9 | 14 |
|
10 | 15 |
|
11 | 16 | # Note: "*_hex" functions are aliases for "(un)hexlify" |
@@ -517,5 +522,63 @@ def test_big_buffer(self, size): |
517 | 522 | self.assertEqual(binascii.crc32(data), 1044521549) |
518 | 523 |
|
519 | 524 |
|
| 525 | +class FreeThreadingTest(unittest.TestCase): |
| 526 | + @unittest.skipUnless(support.Py_GIL_DISABLED, |
| 527 | + 'only meaningful in free-threaded builds') |
| 528 | + def test_import_does_not_enable_gil(self): |
| 529 | + assert_python_ok( |
| 530 | + '-X', 'gil=0', |
| 531 | + '-c', |
| 532 | + ( |
| 533 | + 'import sys\n' |
| 534 | + 'if sys._is_gil_enabled():\n' |
| 535 | + ' raise SystemExit("GIL unexpectedly enabled")\n' |
| 536 | + 'import binascii\n' |
| 537 | + 'if sys._is_gil_enabled():\n' |
| 538 | + ' raise SystemExit("GIL unexpectedly enabled after import")\n' |
| 539 | + ), |
| 540 | + ) |
| 541 | + |
| 542 | + @unittest.skipUnless(support.Py_GIL_DISABLED, |
| 543 | + 'this test can only possibly fail with GIL disabled') |
| 544 | + @threading_helper.reap_threads |
| 545 | + @threading_helper.requires_working_threading() |
| 546 | + def test_free_threading(self): |
| 547 | + if sys._is_gil_enabled(): |
| 548 | + self.skipTest('test requires running with -X gil=0') |
| 549 | + |
| 550 | + num_threads = 8 |
| 551 | + barrier = threading.Barrier(num_threads) |
| 552 | + |
| 553 | + payload = ( |
| 554 | + b'The quick brown fox jumps over the lazy dog.\r\n' |
| 555 | + + bytes(range(256)) |
| 556 | + ) |
| 557 | + hexed = binascii.hexlify(payload) |
| 558 | + b64 = binascii.b2a_base64(payload, newline=False) |
| 559 | + expected_crc = binascii.crc32(payload) |
| 560 | + |
| 561 | + def worker(): |
| 562 | + barrier.wait(timeout=support.SHORT_TIMEOUT) |
| 563 | + for _ in range(1000): |
| 564 | + if binascii.unhexlify(hexed) != payload: |
| 565 | + raise AssertionError('unhexlify mismatch') |
| 566 | + if binascii.hexlify(payload) != hexed: |
| 567 | + raise AssertionError('hexlify mismatch') |
| 568 | + if binascii.a2b_base64(b64) != payload: |
| 569 | + raise AssertionError('a2b_base64 mismatch') |
| 570 | + if binascii.b2a_base64(payload, newline=False) != b64: |
| 571 | + raise AssertionError('b2a_base64 mismatch') |
| 572 | + if binascii.crc32(payload) != expected_crc: |
| 573 | + raise AssertionError('crc32 mismatch') |
| 574 | + |
| 575 | + threads = [threading.Thread(target=worker) for _ in range(num_threads)] |
| 576 | + with threading_helper.catch_threading_exception() as cm: |
| 577 | + with threading_helper.start_threads(threads): |
| 578 | + pass |
| 579 | + if cm.exc_value is not None: |
| 580 | + raise cm.exc_value |
| 581 | + |
| 582 | + |
520 | 583 | if __name__ == "__main__": |
521 | 584 | unittest.main() |
0 commit comments