mirror of
https://github.com/python/cpython.git
synced 2025-10-31 13:41:24 +00:00
42 lines
1.1 KiB
Python
42 lines
1.1 KiB
Python
|
import unittest
|
||
|
|
|
||
|
|
from threading import Thread, Barrier
|
||
|
|
from unittest import TestCase
|
||
|
|
|
||
|
|
from test.support import threading_helper
|
||
|
|
|
||
|
|
|
||
|
|
@threading_helper.requires_working_threading()
|
||
|
|
class TestSet(TestCase):
|
||
|
|
def test_repr_clear(self):
|
||
|
|
"""Test repr() of a set while another thread is calling clear()"""
|
||
|
|
NUM_ITERS = 10
|
||
|
|
NUM_REPR_THREADS = 10
|
||
|
|
barrier = Barrier(NUM_REPR_THREADS + 1)
|
||
|
|
s = {1, 2, 3, 4, 5, 6, 7, 8}
|
||
|
|
|
||
|
|
def clear_set():
|
||
|
|
barrier.wait()
|
||
|
|
s.clear()
|
||
|
|
|
||
|
|
def repr_set():
|
||
|
|
barrier.wait()
|
||
|
|
set_reprs.append(repr(s))
|
||
|
|
|
||
|
|
for _ in range(NUM_ITERS):
|
||
|
|
set_reprs = []
|
||
|
|
threads = [Thread(target=clear_set)]
|
||
|
|
for _ in range(NUM_REPR_THREADS):
|
||
|
|
threads.append(Thread(target=repr_set))
|
||
|
|
for t in threads:
|
||
|
|
t.start()
|
||
|
|
for t in threads:
|
||
|
|
t.join()
|
||
|
|
|
||
|
|
for set_repr in set_reprs:
|
||
|
|
self.assertIn(set_repr, ("set()", "{1, 2, 3, 4, 5, 6, 7, 8}"))
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main()
|