1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
|
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()
|
